zzh
/
gatk-3.8
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Intermediate commit for v3 NanoScheduling algorithm 

-- This version works but it blocks much more than I'd expect on input.  Merging v2 and v3 to make v4 now
This commit is contained in:
Mark DePristo 2012-09-12 10:18:52 -04:00
parent c72db70416
commit 69e418c3f5
9 changed files with 430 additions and 211 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
public/java/src/org/broadinstitute/sting/gatk/traversals/TraverseLociNano.java
View File

@ -21,12 +21,11 @@ import java.util.Iterator;
public class TraverseLociNano<M,T> extends TraverseLociBase<M,T> { public class TraverseLociNano<M,T> extends TraverseLociBase<M,T> {
/** our log, which we want to capture anything from this class */ /** our log, which we want to capture anything from this class */
private static final boolean DEBUG = false; private static final boolean DEBUG = false;
private static final int BUFFER_SIZE = 1000;
final NanoScheduler<MapData, MapResult, T> nanoScheduler; final NanoScheduler<MapData, MapResult, T> nanoScheduler;
public TraverseLociNano(int nThreads) { public TraverseLociNano(int nThreads) {
nanoScheduler = new NanoScheduler<MapData, MapResult, T>(BUFFER_SIZE, nThreads); nanoScheduler = new NanoScheduler<MapData, MapResult, T>(nThreads);
nanoScheduler.setProgressFunction(new TraverseLociProgress()); nanoScheduler.setProgressFunction(new TraverseLociProgress());
} }

4
public/java/src/org/broadinstitute/sting/gatk/traversals/TraverseReadsNano.java
View File

@ -31,7 +31,6 @@ import org.broadinstitute.sting.gatk.datasources.providers.ReadBasedReferenceOrd
import org.broadinstitute.sting.gatk.datasources.providers.ReadReferenceView; import org.broadinstitute.sting.gatk.datasources.providers.ReadReferenceView;
import org.broadinstitute.sting.gatk.datasources.providers.ReadShardDataProvider; import org.broadinstitute.sting.gatk.datasources.providers.ReadShardDataProvider;
import org.broadinstitute.sting.gatk.datasources.providers.ReadView; import org.broadinstitute.sting.gatk.datasources.providers.ReadView;
import org.broadinstitute.sting.gatk.datasources.reads.ReadShard;
import org.broadinstitute.sting.gatk.refdata.RefMetaDataTracker; import org.broadinstitute.sting.gatk.refdata.RefMetaDataTracker;
import org.broadinstitute.sting.gatk.walkers.ReadWalker; import org.broadinstitute.sting.gatk.walkers.ReadWalker;
import org.broadinstitute.sting.utils.GenomeLoc; import org.broadinstitute.sting.utils.GenomeLoc;
@ -60,8 +59,7 @@ public class TraverseReadsNano<M,T> extends TraversalEngine<M,T,ReadWalker<M,T>,
final NanoScheduler<MapData, MapResult, T> nanoScheduler; final NanoScheduler<MapData, MapResult, T> nanoScheduler;
public TraverseReadsNano(int nThreads) { public TraverseReadsNano(int nThreads) {
final int bufferSize = ReadShard.getReadBufferSize() + 1; // actually has 1 more than max nanoScheduler = new NanoScheduler<MapData, MapResult, T>(nThreads);
nanoScheduler = new NanoScheduler<MapData, MapResult, T>(bufferSize, nThreads);
} }
@Override @Override

46
public/java/src/org/broadinstitute/sting/utils/nanoScheduler/InputProducer.java
View File

@ -1,15 +1,13 @@
package org.broadinstitute.sting.utils.nanoScheduler; package org.broadinstitute.sting.utils.nanoScheduler;
import org.broadinstitute.sting.utils.SimpleTimer; import org.broadinstitute.sting.utils.SimpleTimer;
import org.broadinstitute.sting.utils.exceptions.ReviewedStingException;
import java.util.Iterator; import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
/** /**
* Producer Thread that reads input values from an inputReads and puts them into a BlockingQueue * Producer Thread that reads input values from an inputReads and puts them into a BlockingQueue
*/ */
class InputProducer<InputType> implements Runnable { class InputProducer<InputType> {
/** /**
* The iterator we are using to get data from * The iterator we are using to get data from
*/ */
@ -20,42 +18,32 @@ class InputProducer<InputType> implements Runnable {
*/ */
final SimpleTimer inputTimer; final SimpleTimer inputTimer;
/**
* Where we put our input values for consumption
*/
final BlockingQueue<InputValue> outputQueue;
public InputProducer(final Iterator<InputType> inputReader, public InputProducer(final Iterator<InputType> inputReader,
final SimpleTimer inputTimer, final SimpleTimer inputTimer) {
final BlockingQueue<InputValue> outputQueue) {
if ( inputReader == null ) throw new IllegalArgumentException("inputReader cannot be null"); if ( inputReader == null ) throw new IllegalArgumentException("inputReader cannot be null");
if ( inputTimer == null ) throw new IllegalArgumentException("inputTimer cannot be null"); if ( inputTimer == null ) throw new IllegalArgumentException("inputTimer cannot be null");
if ( outputQueue == null ) throw new IllegalArgumentException("OutputQueue cannot be null");
this.inputReader = inputReader; this.inputReader = inputReader;
this.inputTimer = inputTimer; this.inputTimer = inputTimer;
this.outputQueue = outputQueue;
} }
public void run() { public synchronized boolean hasNextNow() {
try { return inputReader.hasNext();
while ( true ) { }
inputTimer.restart();
if ( ! inputReader.hasNext() ) {
inputTimer.stop();
break;
} else {
final InputType input = inputReader.next();
inputTimer.stop();
outputQueue.put(new InputValue(input));
}
}
// add the EOF object so our consumer knows we are done in all inputs public synchronized InputValue next() {
outputQueue.put(new InputValue()); inputTimer.restart();
} catch (InterruptedException ex) {
throw new ReviewedStingException("got execution exception", ex); final InputValue v;
if ( inputReader.hasNext() ) {
v = new InputValue(inputReader.next());
} else {
v = new InputValue();
} }
inputTimer.stop();
return v;
} }
/** /**

13
public/java/src/org/broadinstitute/sting/utils/nanoScheduler/MapResult.java
View File

@ -4,7 +4,7 @@ package org.broadinstitute.sting.utils.nanoScheduler;
* Holds the results of a map job suitable for producer/consumer threading * Holds the results of a map job suitable for producer/consumer threading
* via a BlockingQueue * via a BlockingQueue
*/ */
class MapResult<MapType> extends BlockingQueueValue<MapType> { class MapResult<MapType> extends BlockingQueueValue<MapType> implements Comparable<MapResult<MapType>> {
final int jobID; final int jobID;
/** /**
@ -19,6 +19,12 @@ class MapResult<MapType> extends BlockingQueueValue<MapType> {
if ( jobID < 0 ) throw new IllegalArgumentException("JobID must be >= 0"); if ( jobID < 0 ) throw new IllegalArgumentException("JobID must be >= 0");
} }
MapResult(final int jobID) {
super();
this.jobID = jobID;
if ( jobID < 0 ) throw new IllegalArgumentException("JobID must be >= 0");
}
/** /**
* Create the EOF marker version of MapResult * Create the EOF marker version of MapResult
*/ */
@ -33,4 +39,9 @@ class MapResult<MapType> extends BlockingQueueValue<MapType> {
public int getJobID() { public int getJobID() {
return jobID; return jobID;
} }
@Override
public int compareTo(MapResult<MapType> o) {
return Integer.valueOf(jobID).compareTo(o.getJobID());
}
} }

6
public/java/src/org/broadinstitute/sting/utils/nanoScheduler/NSRuntimeProfile.java
View File

@ -1,7 +1,5 @@
package org.broadinstitute.sting.utils.nanoScheduler; package org.broadinstitute.sting.utils.nanoScheduler;
import com.google.java.contract.Ensures;
import com.google.java.contract.Requires;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import org.broadinstitute.sting.utils.AutoFormattingTime; import org.broadinstitute.sting.utils.AutoFormattingTime;
import org.broadinstitute.sting.utils.SimpleTimer; import org.broadinstitute.sting.utils.SimpleTimer;
@ -46,7 +44,7 @@ public class NSRuntimeProfile {
/** /**
* @return the total runtime for all functions of this nano scheduler * @return the total runtime for all functions of this nano scheduler
*/ */
@Ensures("result >= 0.0") //@Ensures("result >= 0.0")
public double totalRuntimeInSeconds() { public double totalRuntimeInSeconds() {
return inputTimer.getElapsedTime() return inputTimer.getElapsedTime()
+ mapTimer.getElapsedTime() + mapTimer.getElapsedTime()
@ -60,7 +58,7 @@ public class NSRuntimeProfile {
* @param label the name of the timer to display. Should be human readable * @param label the name of the timer to display. Should be human readable
* @param timer the timer whose elapsed time we will display * @param timer the timer whose elapsed time we will display
*/ */
@Requires({"label != null", "timer != null"}) //@Requires({"label != null", "timer != null"})
private void log1(final Logger logger, final String label, final SimpleTimer timer) { private void log1(final Logger logger, final String label, final SimpleTimer timer) {
final double myTimeInSec = timer.getElapsedTime(); final double myTimeInSec = timer.getElapsedTime();
final double myTimePercent = myTimeInSec / totalRuntimeInSeconds() * 100; final double myTimePercent = myTimeInSec / totalRuntimeInSeconds() * 100;

274
public/java/src/org/broadinstitute/sting/utils/nanoScheduler/NanoScheduler.java
View File

@ -1,7 +1,5 @@
package org.broadinstitute.sting.utils.nanoScheduler; package org.broadinstitute.sting.utils.nanoScheduler;
import com.google.java.contract.Ensures;
import com.google.java.contract.Requires;
import org.apache.log4j.Logger; import org.apache.log4j.Logger;
import org.broadinstitute.sting.utils.exceptions.ReviewedStingException; import org.broadinstitute.sting.utils.exceptions.ReviewedStingException;
import org.broadinstitute.sting.utils.threading.NamedThreadFactory; import org.broadinstitute.sting.utils.threading.NamedThreadFactory;
@ -15,13 +13,13 @@ import java.util.concurrent.*;
* *
* The overall framework works like this * The overall framework works like this
* *
* nano <- new Nanoschedule(inputBufferSize, numberOfMapElementsToProcessTogether, nThreads) * nano <- new Nanoschedule(bufferSize, numberOfMapElementsToProcessTogether, nThreads)
* List[Input] outerData : outerDataLoop ) * List[Input] outerData : outerDataLoop )
* result = nano.execute(outerData.iterator(), map, reduce) * result = nano.execute(outerData.iterator(), map, reduce)
* *
* inputBufferSize determines how many elements from the input stream are read in one go by the * bufferSize determines how many elements from the input stream are read in one go by the
* nanoscheduler. The scheduler may hold up to inputBufferSize in memory at one time, as well * nanoscheduler. The scheduler may hold up to bufferSize in memory at one time, as well
* as up to inputBufferSize map results as well. * as up to bufferSize map results as well.
* *
* numberOfMapElementsToProcessTogether determines how many input elements are processed * numberOfMapElementsToProcessTogether determines how many input elements are processed
* together each thread cycle. For example, if this value is 10, then the input data * together each thread cycle. For example, if this value is 10, then the input data
@ -47,12 +45,12 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
private final static int MAP_BUFFER_SIZE_SCALE_FACTOR = 100; private final static int MAP_BUFFER_SIZE_SCALE_FACTOR = 100;
final int inputBufferSize; final int bufferSize;
final int mapBufferSize;
final int nThreads; final int nThreads;
final ExecutorService inputExecutor; final ExecutorService inputExecutor;
final ExecutorService reduceExecutor; final ExecutorService reduceExecutor;
final ThreadPoolExecutor mapExecutor; final ExecutorService mapExecutor;
final Semaphore mapQueueSizeManagingSemaphone;
boolean shutdown = false; boolean shutdown = false;
boolean debug = false; boolean debug = false;
@ -71,24 +69,27 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
/** /**
* Create a new nanoscheduler with the desire characteristics requested by the argument * Create a new nanoscheduler with the desire characteristics requested by the argument
* *
* @param inputBufferSize the number of input elements to read in each scheduling cycle.
* @param nThreads the number of threads to use to get work done, in addition to the * @param nThreads the number of threads to use to get work done, in addition to the
* thread calling execute * thread calling execute
*/ */
public NanoScheduler(final int inputBufferSize, final int nThreads) { public NanoScheduler(final int nThreads) {
if ( inputBufferSize < 1 ) throw new IllegalArgumentException("inputBufferSize must be >= 1, got " + inputBufferSize); this(nThreads*100, nThreads);
}
protected NanoScheduler(final int bufferSize, final int nThreads) {
if ( bufferSize < 1 ) throw new IllegalArgumentException("bufferSize must be >= 1, got " + bufferSize);
if ( nThreads < 1 ) throw new IllegalArgumentException("nThreads must be >= 1, got " + nThreads); if ( nThreads < 1 ) throw new IllegalArgumentException("nThreads must be >= 1, got " + nThreads);
this.inputBufferSize = inputBufferSize; this.bufferSize = bufferSize;
this.mapBufferSize = inputBufferSize * MAP_BUFFER_SIZE_SCALE_FACTOR;
this.nThreads = nThreads; this.nThreads = nThreads;
if ( nThreads == 1 ) { if ( nThreads == 1 ) {
this.mapExecutor = null; this.mapExecutor = this.inputExecutor = this.reduceExecutor = null;
this.inputExecutor = this.reduceExecutor = null; mapQueueSizeManagingSemaphone = null;
} else { } else {
this.mapExecutor = (ThreadPoolExecutor)Executors.newFixedThreadPool(nThreads-1, new NamedThreadFactory("NS-map-thread-%d")); this.mapExecutor = Executors.newFixedThreadPool(nThreads, new NamedThreadFactory("NS-map-thread-%d"));
this.mapExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); mapQueueSizeManagingSemaphone = new Semaphore(this.bufferSize);
this.inputExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("NS-input-thread-%d")); this.inputExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("NS-input-thread-%d"));
this.reduceExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("NS-reduce-thread-%d")); this.reduceExecutor = Executors.newSingleThreadExecutor(new NamedThreadFactory("NS-reduce-thread-%d"));
} }
@ -101,7 +102,7 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
* The number of parallel map threads in use with this NanoScheduler * The number of parallel map threads in use with this NanoScheduler
* @return * @return
*/ */
@Ensures("result > 0") // @Ensures("result > 0")
public int getnThreads() { public int getnThreads() {
return nThreads; return nThreads;
} }
@ -110,9 +111,9 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
* The input buffer size used by this NanoScheduler * The input buffer size used by this NanoScheduler
* @return * @return
*/ */
@Ensures("result > 0") // @Ensures("result > 0")
public int getInputBufferSize() { public int getBufferSize() {
return inputBufferSize; return this.bufferSize;
} }
/** /**
@ -155,8 +156,8 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
* @param name a string name for error messages for the executorService we are shutting down * @param name a string name for error messages for the executorService we are shutting down
* @param executorService the executorService to shut down * @param executorService the executorService to shut down
*/ */
@Requires({"name != null", "executorService != null"}) // @Requires({"name != null", "executorService != null"})
@Ensures("executorService.isShutdown()") // @Ensures("executorService.isShutdown()")
private void shutdownExecutor(final String name, final ExecutorService executorService) { private void shutdownExecutor(final String name, final ExecutorService executorService) {
if ( executorService.isShutdown() || executorService.isTerminated() ) if ( executorService.isShutdown() || executorService.isTerminated() )
throw new IllegalStateException("Executor service " + name + " is already shut down!"); throw new IllegalStateException("Executor service " + name + " is already shut down!");
@ -186,10 +187,10 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
* @param format the format argument suitable for String.format * @param format the format argument suitable for String.format
* @param args the arguments for String.format * @param args the arguments for String.format
*/ */
@Requires("format != null") // @Requires("format != null")
private void debugPrint(final String format, Object ... args) { private void debugPrint(final String format, Object ... args) {
if ( isDebug() ) if ( isDebug() )
logger.info("Thread " + Thread.currentThread().getId() + ":" + String.format(format, args)); logger.warn("Thread " + Thread.currentThread().getId() + ":" + String.format(format, args));
} }
/** /**
@ -261,7 +262,7 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
* *
* @return the reduce result of this map/reduce job * @return the reduce result of this map/reduce job
*/ */
@Requires({"inputReader != null", "map != null", "reduce != null"}) // @Requires({"inputReader != null", "map != null", "reduce != null"})
private ReduceType executeSingleThreaded(final Iterator<InputType> inputReader, private ReduceType executeSingleThreaded(final Iterator<InputType> inputReader,
final NSMapFunction<InputType, MapType> map, final NSMapFunction<InputType, MapType> map,
final ReduceType initialValue, final ReduceType initialValue,
@ -286,7 +287,7 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
if ( LOG_MAP_TIMES ) logger.info("MAP TIME " + (myNSRuntimeProfile.mapTimer.currentTimeNano() - preMapTime)); if ( LOG_MAP_TIMES ) logger.info("MAP TIME " + (myNSRuntimeProfile.mapTimer.currentTimeNano() - preMapTime));
myNSRuntimeProfile.mapTimer.stop(); myNSRuntimeProfile.mapTimer.stop();
if ( i++ % inputBufferSize == 0 && progressFunction != null ) if ( i++ % this.bufferSize == 0 && progressFunction != null )
progressFunction.progress(input); progressFunction.progress(input);
// reduce // reduce
@ -304,89 +305,188 @@ public class NanoScheduler<InputType, MapType, ReduceType> {
* *
* @return the reduce result of this map/reduce job * @return the reduce result of this map/reduce job
*/ */
@Requires({"inputReader != null", "map != null", "reduce != null"}) // @Requires({"inputReader != null", "map != null", "reduce != null"})
private ReduceType executeMultiThreaded(final Iterator<InputType> inputReader, private ReduceType executeMultiThreaded(final Iterator<InputType> inputReader,
final NSMapFunction<InputType, MapType> map, final NSMapFunction<InputType, MapType> map,
final ReduceType initialValue, final ReduceType initialValue,
final NSReduceFunction<MapType, ReduceType> reduce) { final NSReduceFunction<MapType, ReduceType> reduce) {
// debugPrint("Executing nanoScheduler");
//
// // a blocking queue that limits the number of input datum to the requested buffer size
// final BlockingQueue<InputProducer<InputType>.InputValue> inputQueue
// = new LinkedBlockingDeque<InputProducer<InputType>.InputValue>(bufferSize);
//
// // a priority queue that stores up to bufferSize elements
// // produced by completed map jobs.
// final BlockingQueue<Future<MapResult<MapType>>> mapResultQueue =
// new LinkedBlockingDeque<Future<MapResult<MapType>>>(bufferSize);
//
// // Start running the input reader thread
// inputExecutor.submit(new InputProducer<InputType>(inputReader, myNSRuntimeProfile.inputTimer, inputQueue));
//
// // Start running the reducer thread
// final ReducerThread<MapType, ReduceType> reducer
// = new ReducerThread<MapType, ReduceType>(reduce, myNSRuntimeProfile.reduceTimer, initialValue, mapResultQueue);
// final Future<ReduceType> reduceResult = reduceExecutor.submit(reducer);
//
// try {
// int numJobs = 0;
//
// while ( true ) {
// // block on input
// final InputProducer<InputType>.InputValue inputEnqueueWrapped = inputQueue.take();
//
// if ( ! inputEnqueueWrapped.isLast() ) {
// // get the object itself
// final InputType input = inputEnqueueWrapped.getValue();
//
// // the next map call has jobID + 1
// numJobs++;
//
// // send job for map via the completion service
// final CallableMap doMap = new CallableMap(map, numJobs, input);
// final Future<MapResult<MapType>> mapJob = mapExecutor.submit(doMap);
// mapResultQueue.put(mapJob);
//
// debugPrint(" Done with cycle of map/reduce");
//
// if ( numJobs % bufferSize == 0 && progressFunction != null )
// progressFunction.progress(input);
// } else {
// mapResultQueue.put(new FutureValue<MapResult<MapType>>(new MapResult<MapType>()));
// return reduceResult.get(); // wait for our result of reduce
// }
// }
// } catch (InterruptedException ex) {
// throw new ReviewedStingException("got execution exception", ex);
// } catch (ExecutionException ex) {
// throw new ReviewedStingException("got execution exception", ex);
// }
// }
debugPrint("Executing nanoScheduler"); debugPrint("Executing nanoScheduler");
// a blocking queue that limits the number of input datum to the requested buffer size final InputProducer<InputType> inputProducer =
final BlockingQueue<InputProducer<InputType>.InputValue> inputQueue new InputProducer<InputType>(inputReader, myNSRuntimeProfile.inputTimer);
= new LinkedBlockingDeque<InputProducer<InputType>.InputValue>(inputBufferSize);
// a priority queue that stores up to mapBufferSize elements // a priority queue that stores up to bufferSize elements
// produced by completed map jobs. // produced by completed map jobs.
final BlockingQueue<Future<MapResult<MapType>>> mapResultQueue = final PriorityBlockingQueue<MapResult<MapType>> mapResultQueue =
new LinkedBlockingDeque<Future<MapResult<MapType>>>(mapBufferSize); new PriorityBlockingQueue<MapResult<MapType>>();
// Start running the input reader thread final Reducer<MapType, ReduceType> reducer
inputExecutor.submit(new InputProducer<InputType>(inputReader, myNSRuntimeProfile.inputTimer, inputQueue)); = new Reducer<MapType, ReduceType>(reduce, myNSRuntimeProfile.reduceTimer, initialValue);
// Start running the reducer thread
final ReducerThread<MapType, ReduceType> reducer
= new ReducerThread<MapType, ReduceType>(reduce, myNSRuntimeProfile.reduceTimer, initialValue, mapResultQueue);
final Future<ReduceType> reduceResult = reduceExecutor.submit(reducer);
try { try {
int numJobs = 0; int jobID = -1;
while ( true ) { while ( inputProducer.hasNextNow() ) {
// block on input mapQueueSizeManagingSemaphone.acquire();
final InputProducer<InputType>.InputValue inputEnqueueWrapped = inputQueue.take(); jobID++;
debugPrint("Submitting job with id %d", jobID);
if ( ! inputEnqueueWrapped.isLast() ) { mapExecutor.submit(new ReadMapReduceJob(jobID, inputProducer, mapResultQueue, map, reducer));
// get the object itself
final InputType input = inputEnqueueWrapped.getValue();
// the next map call has jobID + 1
numJobs++;
// send job for map via the completion service
final CallableMap doMap = new CallableMap(map, numJobs, input);
final Future<MapResult<MapType>> mapJob = mapExecutor.submit(doMap);
mapResultQueue.put(mapJob);
debugPrint(" Done with cycle of map/reduce");
if ( numJobs % inputBufferSize == 0 && progressFunction != null )
progressFunction.progress(input);
} else {
mapResultQueue.put(new FutureValue<MapResult<MapType>>(new MapResult<MapType>()));
return reduceResult.get(); // wait for our result of reduce
}
} }
debugPrint("Setting last job id %d", jobID);
reducer.setLastJobID(jobID); // the last actually submitted job id is jobID - 1
return reducer.waitForFinalReduce();
} catch (InterruptedException ex) { } catch (InterruptedException ex) {
throw new ReviewedStingException("got execution exception", ex); throw new ReviewedStingException("got execution exception", ex);
} catch (ExecutionException ex) { // } catch (ExecutionException ex) {
throw new ReviewedStingException("got execution exception", ex); // throw new ReviewedStingException("got execution exception", ex);
} }
} }
/** private class ReadMapReduceJob implements Runnable {
* A simple callable version of the map function for use with the executor pool final int jobID;
*/ final InputProducer<InputType> inputProducer;
private class CallableMap implements Callable<MapResult<MapType>> { final BlockingQueue<MapResult<MapType>> mapResultQueue;
final int id;
final InputType input;
final NSMapFunction<InputType, MapType> map; final NSMapFunction<InputType, MapType> map;
final Reducer<MapType, ReduceType> reducer;
@Requires({"map != null"}) private ReadMapReduceJob(final int jobID,
private CallableMap(final NSMapFunction<InputType, MapType> map, final InputProducer<InputType> inputProducer,
final int id, final BlockingQueue<MapResult<MapType>> mapResultQueue,
final InputType input) { final NSMapFunction<InputType, MapType> map,
this.id = id; final Reducer<MapType, ReduceType> reducer) {
this.input = input; this.jobID = jobID;
this.inputProducer = inputProducer;
this.mapResultQueue = mapResultQueue;
this.map = map; this.map = map;
this.reducer = reducer;
} }
@Override @Override
public MapResult<MapType> call() { public void run() {
if ( debug ) debugPrint("\t\tmap " + input); try {
myNSRuntimeProfile.mapTimer.restart(); debugPrint("Running ReadMapReduceJob " + jobID);
final MapType result = map.apply(input); final InputProducer<InputType>.InputValue inputWrapper = inputProducer.next();
myNSRuntimeProfile.mapTimer.stop();
return new MapResult<MapType>(result, id); final MapResult<MapType> result;
if ( ! inputWrapper.isLast() ) {
// just skip doing anything if we don't have work to do, which is possible
// because we don't necessarily know how much input there is when we queue
// up our jobs
final InputType input = inputWrapper.getValue();
// map
myNSRuntimeProfile.mapTimer.restart();
final long preMapTime = LOG_MAP_TIMES ? 0 : myNSRuntimeProfile.mapTimer.currentTimeNano();
final MapType mapValue = map.apply(input);
if ( LOG_MAP_TIMES ) logger.info("MAP TIME " + (myNSRuntimeProfile.mapTimer.currentTimeNano() - preMapTime));
myNSRuntimeProfile.mapTimer.stop();
// enqueue the result into the mapResultQueue
result = new MapResult<MapType>(mapValue, jobID);
if ( jobID % bufferSize == 0 && progressFunction != null )
progressFunction.progress(input);
} else {
// if there's no input we push empty MapResults with jobIDs for synchronization with Reducer
result = new MapResult<MapType>(jobID);
}
mapResultQueue.put(result);
debugPrint(" Pushed MapResult with job id %d", jobID);
final int nReduced = reducer.reduceAsMuchAsPossible(mapResultQueue);
debugPrint(" reduced %d values", nReduced);
// we finished a map job, release the job queue semaphore
mapQueueSizeManagingSemaphone.release();
} catch (InterruptedException ex) {
throw new ReviewedStingException("got execution exception", ex);
// } catch (ExecutionException ex) {
// throw new ReviewedStingException("got execution exception", ex);
}
} }
} }
// /**
// * A simple callable version of the map function for use with the executor pool
// */
// private class CallableMap implements Callable<MapResult<MapType>> {
// final int id;
// final InputType input;
// final NSMapFunction<InputType, MapType> map;
//
// @Requires({"map != null"})
// private CallableMap(final NSMapFunction<InputType, MapType> map,
// final int id,
// final InputType input) {
// this.id = id;
// this.input = input;
// this.map = map;
// }
//
// @Override
// public MapResult<MapType> call() {
// if ( debug ) debugPrint("\t\tmap " + input);
// myNSRuntimeProfile.mapTimer.restart();
// final MapType result = map.apply(input);
// myNSRuntimeProfile.mapTimer.stop();
// return new MapResult<MapType>(result, id);
// }
// }
} }

88
public/java/src/org/broadinstitute/sting/utils/nanoScheduler/Reducer.java 100644
View File

@ -0,0 +1,88 @@
package org.broadinstitute.sting.utils.nanoScheduler;
import org.broadinstitute.sting.utils.SimpleTimer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
/**
* Thread that runs the reduce of the map/reduce.
*
* This thread reads from mapResultsQueue until the poison EOF object arrives. At each
* stage is calls reduce(value, sum). The blocking mapResultQueue ensures that the
* queue waits until the mapResultQueue has a value to take. Then, it gets and waits
* until the map result Future has a value.
*/
class Reducer<MapType, ReduceType> {
final CountDownLatch countDownLatch = new CountDownLatch(1);
final NSReduceFunction<MapType, ReduceType> reduce;
final SimpleTimer reduceTimer;
ReduceType sum;
int lastJobID = -2; // not yet set
int prevJobID = -1; // no jobs observed
public Reducer(final NSReduceFunction<MapType, ReduceType> reduce,
final SimpleTimer reduceTimer,
final ReduceType initialSum) {
if ( reduce == null ) throw new IllegalArgumentException("Reduce function cannot be null");
if ( reduceTimer == null ) throw new IllegalArgumentException("reduceTimer cannot be null");
this.reduce = reduce;
this.reduceTimer = reduceTimer;
this.sum = initialSum;
}
private synchronized boolean readyToReduce(final BlockingQueue<MapResult<MapType>> mapResultQueue) {
final MapResult<MapType> nextMapResult = mapResultQueue.peek();
return nextMapResult != null && nextMapResult.getJobID() == prevJobID + 1;
}
public synchronized int reduceAsMuchAsPossible(final BlockingQueue<MapResult<MapType>> mapResultQueue) throws InterruptedException {
int nReduces = 0;
while ( readyToReduce(mapResultQueue) ) {
final MapResult<MapType> result = mapResultQueue.take();
if ( result.getJobID() < prevJobID )
// make sure the map results are coming in order
throw new IllegalStateException("BUG: last jobID " + prevJobID + " > current jobID " + result.getJobID());
prevJobID = result.getJobID();
if ( ! result.isLast() ) { // TODO -- rename to isEmpty
nReduces++;
// apply reduce, keeping track of sum
reduceTimer.restart();
sum = reduce.apply(result.getValue(), sum);
reduceTimer.stop();
}
maybeReleaseLatch();
}
return nReduces;
}
private synchronized void maybeReleaseLatch() {
if ( lastJobID != -2 && (prevJobID == lastJobID || lastJobID == -1) ) {
// either we've already seen the last one prevJobID == lastJobID or
// the last job ID is -1, meaning that no jobs were ever submitted
countDownLatch.countDown();
}
}
public synchronized void setLastJobID(final int lastJobID) {
if ( lastJobID < -1 ) throw new IllegalArgumentException("lastJobID must be > -1, but saw " + lastJobID);
this.lastJobID = lastJobID;
maybeReleaseLatch();
}
public ReduceType waitForFinalReduce() throws InterruptedException {
countDownLatch.await();
return sum;
}
}

144
public/java/test/org/broadinstitute/sting/utils/nanoScheduler/InputProducerUnitTest.java
View File

@ -1,72 +1,72 @@
package org.broadinstitute.sting.utils.nanoScheduler; //package org.broadinstitute.sting.utils.nanoScheduler;
//
import org.broadinstitute.sting.BaseTest; //import org.broadinstitute.sting.BaseTest;
import org.broadinstitute.sting.utils.SimpleTimer; //import org.broadinstitute.sting.utils.SimpleTimer;
import org.testng.Assert; //import org.testng.Assert;
import org.testng.annotations.DataProvider; //import org.testng.annotations.DataProvider;
import org.testng.annotations.Test; //import org.testng.annotations.Test;
//
import java.util.ArrayList; //import java.util.ArrayList;
import java.util.Arrays; //import java.util.Arrays;
import java.util.List; //import java.util.List;
import java.util.concurrent.ExecutorService; //import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; //import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque; //import java.util.concurrent.LinkedBlockingDeque;
//
/** ///**
* UnitTests for the InputProducer // * UnitTests for the InputProducer
* // *
* User: depristo // * User: depristo
* Date: 8/24/12 // * Date: 8/24/12
* Time: 11:25 AM // * Time: 11:25 AM
* To change this template use File | Settings | File Templates. // * To change this template use File | Settings | File Templates.
*/ // */
public class InputProducerUnitTest extends BaseTest { //public class InputProducerUnitTest extends BaseTest {
@DataProvider(name = "InputProducerTest") // @DataProvider(name = "InputProducerTest")
public Object[][] createInputProducerTest() { // public Object[][] createInputProducerTest() {
List<Object[]> tests = new ArrayList<Object[]>(); // List<Object[]> tests = new ArrayList<Object[]>();
//
for ( final int nElements : Arrays.asList(0, 1, 10, 100, 1000, 10000, 100000) ) { // for ( final int nElements : Arrays.asList(0, 1, 10, 100, 1000, 10000, 100000) ) {
for ( final int queueSize : Arrays.asList(1, 10, 100) ) { // for ( final int queueSize : Arrays.asList(1, 10, 100) ) {
tests.add(new Object[]{ nElements, queueSize }); // tests.add(new Object[]{ nElements, queueSize });
} // }
} // }
//
return tests.toArray(new Object[][]{}); // return tests.toArray(new Object[][]{});
} // }
//
@Test(enabled = true, dataProvider = "InputProducerTest", timeOut = NanoSchedulerUnitTest.NANO_SCHEDULE_MAX_RUNTIME) // @Test(enabled = true, dataProvider = "InputProducerTest", timeOut = NanoSchedulerUnitTest.NANO_SCHEDULE_MAX_RUNTIME)
public void testInputProducer(final int nElements, final int queueSize) throws InterruptedException { // public void testInputProducer(final int nElements, final int queueSize) throws InterruptedException {
final List<Integer> elements = new ArrayList<Integer>(nElements); // final List<Integer> elements = new ArrayList<Integer>(nElements);
for ( int i = 0; i < nElements; i++ ) elements.add(i); // for ( int i = 0; i < nElements; i++ ) elements.add(i);
//
final LinkedBlockingDeque<InputProducer<Integer>.InputValue> readQueue = // final LinkedBlockingDeque<InputProducer<Integer>.InputValue> readQueue =
new LinkedBlockingDeque<InputProducer<Integer>.InputValue>(queueSize); // new LinkedBlockingDeque<InputProducer<Integer>.InputValue>(queueSize);
//
final InputProducer<Integer> ip = new InputProducer<Integer>(elements.iterator(), new SimpleTimer(), readQueue); // final InputProducer<Integer> ip = new InputProducer<Integer>(elements.iterator(), new SimpleTimer(), readQueue);
//
final ExecutorService es = Executors.newSingleThreadExecutor(); // final ExecutorService es = Executors.newSingleThreadExecutor();
es.submit(ip); // es.submit(ip);
//
int lastValue = -1; // int lastValue = -1;
int nRead = 0; // int nRead = 0;
while ( true ) { // while ( true ) {
final int observedQueueSize = readQueue.size(); // final int observedQueueSize = readQueue.size();
Assert.assertTrue(observedQueueSize <= queueSize, // Assert.assertTrue(observedQueueSize <= queueSize,
"Reader is enqueuing more elements " + observedQueueSize + " than allowed " + queueSize); // "Reader is enqueuing more elements " + observedQueueSize + " than allowed " + queueSize);
//
final InputProducer<Integer>.InputValue value = readQueue.take(); // final InputProducer<Integer>.InputValue value = readQueue.take();
if ( value.isLast() ) { // if ( value.isLast() ) {
Assert.assertEquals(nRead, nElements, "Number of input values " + nRead + " not all that are expected " + nElements); // Assert.assertEquals(nRead, nElements, "Number of input values " + nRead + " not all that are expected " + nElements);
Assert.assertEquals(readQueue.size(), 0, "Last queue element found but queue contains more values!"); // Assert.assertEquals(readQueue.size(), 0, "Last queue element found but queue contains more values!");
break; // break;
} else { // } else {
Assert.assertTrue(lastValue < value.getValue(), "Read values coming out of order!"); // Assert.assertTrue(lastValue < value.getValue(), "Read values coming out of order!");
final int expected = lastValue + 1; // final int expected = lastValue + 1;
Assert.assertEquals((int)value.getValue(), expected, "Value observed " + value.getValue() + " not equal to the expected value " + expected); // Assert.assertEquals((int)value.getValue(), expected, "Value observed " + value.getValue() + " not equal to the expected value " + expected);
nRead++; // nRead++;
lastValue = value.getValue(); // lastValue = value.getValue();
} // }
} // }
} // }
} //}

63
public/java/test/org/broadinstitute/sting/utils/nanoScheduler/NanoSchedulerUnitTest.java
View File

@ -27,6 +27,22 @@ public class NanoSchedulerUnitTest extends BaseTest {
@Override public Integer apply(Integer input) { return input * 2; } @Override public Integer apply(Integer input) { return input * 2; }
} }
private static class Map2xWithDelays extends Map2x {
@Override public Integer apply(Integer input) {
try {
if ( input % 7 == 0 ) {
final int milliToSleep = (input % 10);
//System.out.printf("Sleeping %d millseconds%n", milliToSleep);
Thread.sleep(milliToSleep);
}
return input * 2;
} catch ( InterruptedException ex ) {
throw new RuntimeException(ex);
}
}
}
private static class ReduceSum implements NSReduceFunction<Integer, Integer> { private static class ReduceSum implements NSReduceFunction<Integer, Integer> {
int prevOne = Integer.MIN_VALUE; int prevOne = Integer.MIN_VALUE;
@ -55,16 +71,18 @@ public class NanoSchedulerUnitTest extends BaseTest {
private static class NanoSchedulerBasicTest extends TestDataProvider { private static class NanoSchedulerBasicTest extends TestDataProvider {
final int bufferSize, nThreads, start, end, expectedResult; final int bufferSize, nThreads, start, end, expectedResult;
final boolean addDelays;
public NanoSchedulerBasicTest(final int bufferSize, final int nThreads, final int start, final int end) { public NanoSchedulerBasicTest(final int bufferSize, final int nThreads, final int start, final int end, final boolean addDelays) {
super(NanoSchedulerBasicTest.class); super(NanoSchedulerBasicTest.class);
this.bufferSize = bufferSize; this.bufferSize = bufferSize;
this.nThreads = nThreads; this.nThreads = nThreads;
this.start = start; this.start = start;
this.end = end; this.end = end;
this.expectedResult = sum2x(start, end); this.expectedResult = sum2x(start, end);
setName(String.format("%s nt=%d buf=%d start=%d end=%d sum=%d", this.addDelays = addDelays;
getClass().getSimpleName(), nThreads, bufferSize, start, end, expectedResult)); setName(String.format("%s nt=%d buf=%d start=%d end=%d sum=%d delays=%b",
getClass().getSimpleName(), nThreads, bufferSize, start, end, expectedResult, addDelays));
} }
public Iterator<Integer> makeReader() { public Iterator<Integer> makeReader() {
@ -79,19 +97,39 @@ public class NanoSchedulerUnitTest extends BaseTest {
return nElements / bufferSize; return nElements / bufferSize;
} }
public Map2x makeMap() { return new Map2x(); } public Map2x makeMap() { return addDelays ? new Map2xWithDelays() : new Map2x(); }
public Integer initReduce() { return 0; } public Integer initReduce() { return 0; }
public ReduceSum makeReduce() { return new ReduceSum(); } public ReduceSum makeReduce() { return new ReduceSum(); }
public NanoScheduler<Integer, Integer, Integer> makeScheduler() {
if ( bufferSize == -1 )
return new NanoScheduler<Integer, Integer, Integer>(nThreads);
else
return new NanoScheduler<Integer, Integer, Integer>(bufferSize, nThreads);
}
} }
static NanoSchedulerBasicTest exampleTest = null; static NanoSchedulerBasicTest exampleTest = null;
@DataProvider(name = "NanoSchedulerBasicTest") @DataProvider(name = "NanoSchedulerBasicTest")
public Object[][] createNanoSchedulerBasicTest() { public Object[][] createNanoSchedulerBasicTest() {
for ( final int bufferSize : Arrays.asList(1, 10, 1000, 1000000, 10000000) ) { // for ( final int bufferSize : Arrays.asList(1, 10) ) {
// for ( final int nt : Arrays.asList(1, 2, 4) ) {
// for ( final int start : Arrays.asList(0) ) {
// for ( final int end : Arrays.asList(0, 1, 2) ) {
// exampleTest = new NanoSchedulerBasicTest(bufferSize, nt, start, end);
// }
// }
// }
// }
for ( final int bufferSize : Arrays.asList(-1, 1, 10, 100) ) {
for ( final int nt : Arrays.asList(1, 2, 4) ) { for ( final int nt : Arrays.asList(1, 2, 4) ) {
for ( final int start : Arrays.asList(0) ) { for ( final int start : Arrays.asList(0) ) {
for ( final int end : Arrays.asList(0, 1, 2, 11, 10000, 100000) ) { for ( final int end : Arrays.asList(0, 1, 2, 11, 100, 10000, 100000) ) {
exampleTest = new NanoSchedulerBasicTest(bufferSize, nt, start, end); for ( final boolean addDelays : Arrays.asList(true, false) ) {
if ( end < 1000 )
exampleTest = new NanoSchedulerBasicTest(bufferSize, nt, start, end, addDelays);
}
} }
} }
} }
@ -116,13 +154,13 @@ public class NanoSchedulerUnitTest extends BaseTest {
private void testNanoScheduler(final NanoSchedulerBasicTest test) throws InterruptedException { private void testNanoScheduler(final NanoSchedulerBasicTest test) throws InterruptedException {
final SimpleTimer timer = new SimpleTimer().start(); final SimpleTimer timer = new SimpleTimer().start();
final NanoScheduler<Integer, Integer, Integer> nanoScheduler = final NanoScheduler<Integer, Integer, Integer> nanoScheduler = test.makeScheduler();
new NanoScheduler<Integer, Integer, Integer>(test.bufferSize, test.nThreads);
final ProgressCallback callback = new ProgressCallback(); final ProgressCallback callback = new ProgressCallback();
nanoScheduler.setProgressFunction(callback); nanoScheduler.setProgressFunction(callback);
Assert.assertEquals(nanoScheduler.getInputBufferSize(), test.bufferSize, "inputBufferSize argument"); if ( test.bufferSize > -1 )
Assert.assertEquals(nanoScheduler.getBufferSize(), test.bufferSize, "bufferSize argument");
Assert.assertEquals(nanoScheduler.getnThreads(), test.nThreads, "nThreads argument"); Assert.assertEquals(nanoScheduler.getnThreads(), test.nThreads, "nThreads argument");
final Integer sum = nanoScheduler.execute(test.makeReader(), test.makeMap(), test.initReduce(), test.makeReduce()); final Integer sum = nanoScheduler.execute(test.makeReader(), test.makeMap(), test.initReduce(), test.makeReduce());
@ -149,8 +187,7 @@ public class NanoSchedulerUnitTest extends BaseTest {
if ( test.bufferSize > 1) { if ( test.bufferSize > 1) {
logger.warn("Running " + test); logger.warn("Running " + test);
final NanoScheduler<Integer, Integer, Integer> nanoScheduler = final NanoScheduler<Integer, Integer, Integer> nanoScheduler = test.makeScheduler();
new NanoScheduler<Integer, Integer, Integer>(test.bufferSize, test.nThreads);
// test reusing the scheduler // test reusing the scheduler
for ( int i = 0; i < 10; i++ ) { for ( int i = 0; i < 10; i++ ) {
@ -183,7 +220,7 @@ public class NanoSchedulerUnitTest extends BaseTest {
BasicConfigurator.configure(); BasicConfigurator.configure();
logger.setLevel(org.apache.log4j.Level.DEBUG); logger.setLevel(org.apache.log4j.Level.DEBUG);
final NanoSchedulerBasicTest test = new NanoSchedulerBasicTest(1000, Integer.valueOf(args[0]), 0, Integer.valueOf(args[1])); final NanoSchedulerBasicTest test = new NanoSchedulerBasicTest(1000, Integer.valueOf(args[0]), 0, Integer.valueOf(args[1]), false);
final NanoScheduler<Integer, Integer, Integer> nanoScheduler = final NanoScheduler<Integer, Integer, Integer> nanoScheduler =
new NanoScheduler<Integer, Integer, Integer>(test.bufferSize, test.nThreads); new NanoScheduler<Integer, Integer, Integer>(test.bufferSize, test.nThreads);
nanoScheduler.setDebug(true); nanoScheduler.setDebug(true);