/**
 * Calculate edit distance from targetText to known word.
 *
 * @param word known word
 * @param v0 int array of length targetText.length() + 1
 * @param v1 int array of length targetText.length() + 1
 * @return distance
 */
private int editDistance(String word, int[] v0, int[] v1) {
    
    // initialize v0 (prior row of distances) as edit distance for empty 'word'
    for (int i = 0; i < v0.length; i++) {
        v0[i] = i;
    }
    
    // calculate updated v0 (current row distances) from the previous row v0
    for (int i = 0; i < word.length(); i++) {
        
        // first element of v1 = delete (i+1) chars from target to match empty 'word'
        v1[0] = i + 1;
        
        // use formula to fill in the rest of the row
        for (int j = 0; j < targetText.length(); j++) {
            int cost = (word.charAt(i) == targetText.charAt(j)) ? 0 : 1;
            v1[j + 1] = minimum(v1[j] + 1, v0[j + 1] + 1, v0[j] + cost);
        }
        
        // swap v1 (current row) and v0 (previous row) for next iteration
        int[] hold = v0;
        v0 = v1;
        v1 = hold;
    }
    
    // return final value representing best edit distance
    return v0[targetText.length()];
}

private final ExecutorService threadPool;
private final String[] knownWords;
private final int blockSize;

public ThreadPoolDistance(String[] words, int block) {
threadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
knownWords = words;
blockSize = block;
}

public DistancePair bestMatch(String target) {

// build a list of tasks for matching to ranges of known words
List<DistanceTask> tasks = new ArrayList<DistanceTask>();

int size = 0;
for (int base = 0; base < knownWords.length; base += size) {
size = Math.min(blockSize, knownWords.length - base);
tasks.add(new DistanceTask(target, base, size));
}
DistancePair best;
try {

// pass the list of tasks to the executor, getting back list of futures
List<Future<DistancePair>> results = threadPool.invokeAll(tasks);

// find the best result, waiting for each future to complete
best = DistancePair.WORST_CASE;
for (Future<DistancePair> future: results) {
DistancePair result = future.get();
best = DistancePair.best(best, result);
}

} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
return best;
}

/**
* Shortest distance task implementation using Callable.
*/
public class DistanceTask implements Callable<DistancePair>
{
private final String targetText;
private final int startOffset;
private final int compareCount;

public DistanceTask(String target, int offset, int count) {
targetText = target;
startOffset = offset;
compareCount = count;
}

private int editDistance(String word, int[] v0, int[] v1) {
...
}

/* (non-Javadoc)
* @see java.util.concurrent.Callable#call()
*/
@Override
public DistancePair call() throws Exception {

// directly compare distances for comparison words in range
int[] v0 = new int[targetText.length() + 1];
int[] v1 = new int[targetText.length() + 1];
int bestIndex = -1;
int bestDistance = Integer.MAX_VALUE;
boolean single = false;
for (int i = 0; i < compareCount; i++) {
int distance = editDistance(knownWords[i + startOffset], v0, v1);
if (bestDistance > distance) {
bestDistance = distance;
bestIndex = i + startOffset;
single = true;
} else if (bestDistance == distance) {
single = false;
}
}
return single ? new DistancePair(bestDistance, knownWords[bestIndex]) :
new DistancePair(bestDistance);
}
}

private ForkJoinPool threadPool = new ForkJoinPool();

private final String[] knownWords;

private final int blockSize;

public ForkJoinDistance(String[] words, int block) {
knownWords = words;
blockSize = block;
}

public DistancePair bestMatch(String target) {
return threadPool.invoke(new DistanceTask(target, 0, knownWords.length, knownWords));
}

/**
* Shortest distance task implementation using RecursiveTask.
*/
public class DistanceTask extends RecursiveTask<DistancePair>
{
private final String compareText;
private final int startOffset;
private final int compareCount;
private final String[] matchWords;

public DistanceTask(String from, int offset, int count, String[] words) {
compareText = from;
startOffset = offset;
compareCount = count;
matchWords = words;
}

private int editDistance(int index, int[] v0, int[] v1) {
...
}

/* (non-Javadoc)
* @see java.util.concurrent.RecursiveTask#compute()
*/
@Override
protected DistancePair compute() {
if (compareCount > blockSize) {

// split range in half and find best result from bests in each half of range
int half = compareCount / 2;
DistanceTask t1 = new DistanceTask(compareText, startOffset, half, matchWords);
t1.fork();
DistanceTask t2 = new DistanceTask(compareText, startOffset + half,
compareCount - half, matchWords);
DistancePair p2 = t2.compute();
return DistancePair.best(p2, t1.join());
}

// directly compare distances for comparison words in range
int[] v0 = new int[compareText.length() + 1];
int[] v1 = new int[compareText.length() + 1];
int bestIndex = -1;
int bestDistance = Integer.MAX_VALUE;
boolean single = false;
for (int i = 0; i < compareCount; i++) {
int distance = editDistance(i + startOffset, v0, v1);
if (bestDistance > distance) {
bestDistance = distance;
bestIndex = i + startOffset;
single = true;
} else if (bestDistance == distance) {
single = false;
}
}
return single ? new DistancePair(bestDistance, knownWords[bestIndex]) :
new DistancePair(bestDistance);
}
}

import ExecutionContext.Implicits.global

val lastInteger = new AtomicInteger
def futureInt() = future {
Thread sleep 2000
lastInteger incrementAndGet
}

// use callbacks for completion of futures
val a1 = futureInt
val a2 = futureInt
a1.onSuccess {
case i1 => {
a2.onSuccess {
case i2 => println("Sum of values is " + (i1 + i2))
}
}
}
Thread sleep 3000

// use for construct to extract values when futures complete
val b1 = futureInt
val b2 = futureInt
for (i1 <- b1; i2 <- b2) yield println("Sum of values is " + (i1 + i2))
Thread sleep 3000

// wait directly for completion of futures
val c1 = futureInt
val c2 = futureInt
println("Sum of values is " + (Await.result(c1, Duration.Inf) +
Await.result(c2, Duration.Inf)))

val limit = targetText.length
/** Calculate edit distance from targetText to known word.
  *
  * @param word known word
  * @param v0 int array of length targetText.length + 1
  * @param v1 int array of length targetText.length + 1
  * @return distance
  */
def editDistance(word: String, v0: Array[Int], v1: Array[Int]) = {

val length = word.length

@tailrec
def distanceByRow(rnum: Int, r0: Array[Int], r1: Array[Int]): Int = {
if (rnum >= length) r0(limit)
else {

// first element of r1 = delete (i+1) chars from target to match empty 'word'
r1(0) = rnum + 1

// use formula to fill in the rest of the row
for (j <- 0 until limit) {
val cost = if (word(rnum) == targetText(j)) 0 else 1
r1(j + 1) = min(r1(j) + 1, r0(j + 1) + 1, r0(j) + cost);
}

// recurse with arrays swapped for next row
distanceByRow(rnum + 1, r1, r0)
}
}

// initialize v0 (prior row of distances) as edit distance for empty 'word'
for (i <- 0 to limit) v0(i) = i

// recursively process rows matching characters in word being compared to find best
distanceByRow(0, v0, v1)
}

val limit = targetText.length

/** Calculate edit distance from targetText to known word.
*
* @param word known word
* @param v0 int array of length targetText.length + 1
* @param v1 int array of length targetText.length + 1
* @return distance
*/
def editDistance(word: String, v0: Array[Int], v1: Array[Int]) = {

val length = word.length

@tailrec
def distanceByRow(row: Int, r0: Array[Int], r1: Array[Int]): Int = {
if (row >= length) r0(limit)
else {

// first element of v1 = delete (i+1) chars from target to match empty 'word'
r1(0) = row + 1

// use formula recursively to fill in the rest of the row
@tailrec
def distanceByColumn(col: Int): Unit = {
if (col < limit) {
val cost = if (word(row) == targetText(col)) 0 else 1
r1(col + 1) = min(r1(col) + 1, r0(col + 1) + 1, r0(col) + cost)
distanceByColumn(col + 1)
}
}
distanceByColumn(0)

// recurse with arrays swapped for next row
distanceByRow(row + 1, r1, r0)
}
}

// initialize v0 (prior row of distances) as edit distance for empty 'word'
@tailrec
def initArray(index: Int): Unit = {
if (index <= limit) {
v0(index) = index
initArray(index + 1)
}
}
initArray(0)

// recursively process rows matching characters in word being compared to find best
distanceByRow(0, v0, v1)
}

class Matcher(words: Array[String]) {

def bestMatch(targetText: String) = {

val limit = targetText.length
val v0 = new Array[Int](limit + 1)
val v1 = new Array[Int](limit + 1)

def editDistance(word: String, v0: Array[Int], v1: Array[Int]) = {
...
}

@tailrec
/** Scan all known words in range to find best match.
*
* @param index next word index
* @param bestDist minimum distance found so far
* @param bestMatch unique word at minimum distance, or None if not unique
* @return best match
*/
def best(index: Int, bestDist: Int, bestMatch: Option[String]): DistancePair =
if (index < words.length) {
val newDist = editDistance(words(index), v0, v1)
val next = index + 1
if (newDist < bestDist) best(next, newDist, Some(words(index)))
else if (newDist == bestDist) best(next, bestDist, None)
else best(next, bestDist, bestMatch)
} else DistancePair(bestDist, bestMatch)

best(0, Int.MaxValue, None)
}
}

class ParallelCollectionDistance(words: Array[String], size: Int) extends TimingTestBase {

val matchers = words.grouped(size).map(l => new Matcher(l)).toList

def shutdown = {}

def blockSize = size

/** Find best result across all matchers, using parallel collection. */
def bestMatch(target: String) = {
matchers.par.map(m => m.bestMatch(target)).
foldLeft(DistancePair.worstMatch)((a, m) => DistancePair.best(a, m))
}
}

class DirectBlockingDistance(words: Array[String], size: Int) extends TimingTestBase {

val matchers = words.grouped(size).map(l => new Matcher(l)).toList

def shutdown = {}

def blockSize = size

/** Find best result across all matchers, using direct blocking waits. */
def bestMatch(target: String) = {
import ExecutionContext.Implicits.global
val futures = matchers.map(m => future { m.bestMatch(target) })
futures.foldLeft(DistancePair.worstMatch)((a, v) =>
DistancePair.best(a, Await.result(v, Duration.Inf)))
}
}