Commit ff99ef90 authored by Matija Obreza's avatar Matija Obreza
Browse files

Buffered input reader and larger batch size

parent c1074f2c
......@@ -20,6 +20,7 @@ import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
......@@ -85,13 +86,14 @@ public class CountryCLI {
int columnOrigCty = 5;
CSVWriter writer = new CSVWriter(new OutputStreamWriter(System.out), ',', '"', '\\', "\n");
CSVReader reader = new CSVReader(new InputStreamReader(System.in), ',', '"', '\\', 0, false);
CSVReader reader = new CSVReader(new BufferedReader(new InputStreamReader(System.in), 10000), ',', '"', '\\',
0, false);
CountryOfOriginServiceImpl countryOfOriginService = new CountryOfOriginServiceImpl();
countryOfOriginService.afterPropertiesSet();
// Thread pool with nThreads concurrent threads
int nThreads = 4;
int nThreads = 7;
ExecutorService threadPool = Executors.newFixedThreadPool(nThreads);
StopWatch stopWatch = new StopWatch();
......@@ -102,77 +104,109 @@ public class CountryCLI {
long lastLogTime = 0;
stopWatch.start();
String[][] inputLines = new String[nThreads][];
ArrayList<?>[] inputLines = new ArrayList[nThreads];
for (int i = 0; i < nThreads; i++) {
inputLines[i] = new ArrayList<String[]>();
}
Future<?>[] futures = new Future<?>[nThreads];
do {
submitted = 0;
// Reset
for (int i = nThreads - 1; i >= 0; i--) {
inputLines[i] = null;
for (int i = 0; i < nThreads; i++) {
inputLines[i].clear();
futures[i] = null;
submitted = 0;
}
// Submit
for (int i = nThreads - 1; i >= 0; i--) {
String[] nextLine = reader.readNext();
lineCount++;
stopWatch.split();
long processingTimeSeconds = stopWatch.getSplitTime() / 1000;
if ((lineCount % 1000 == 0)
|| (lastLogTime != processingTimeSeconds && processingTimeSeconds % 10 == 0)) {
System.err.println("FYI, " + lineCount + " entries have been processed in " + processingTimeSeconds
+ " = " + (lineCount / processingTimeSeconds) + " lines/second");
countryOfOriginService.printCache();
lastLogTime = processingTimeSeconds;
for (int i = 0; i < nThreads; i++) {
final ArrayList<String[]> inputs = (ArrayList<String[]>) inputLines[i];
for (int j = 0; j < 100; j++) {
String[] nextLine = reader.readNext();
lineCount++;
stopWatch.split();
long processingTimeSeconds = stopWatch.getSplitTime() / 1000;
if ((lineCount % 1000 == 0)
|| (lastLogTime != processingTimeSeconds && processingTimeSeconds % 10 == 0)) {
System.err
.println("FYI, " + lineCount + " entries have been processed in "
+ processingTimeSeconds + " = " + (lineCount / processingTimeSeconds)
+ " lines/second");
countryOfOriginService.printCache();
lastLogTime = processingTimeSeconds;
}
if (nextLine == null) {
System.err.println("No input");
break;
}
inputs.add(nextLine);
}
if (nextLine == null)
if (inputs.size() == 0)
break;
inputLines[i] = nextLine;
submitted++;
// final int cpuId = i;
// System.err.println("Submitted " + cpuId);
futures[i] = threadPool.submit(new Callable<String>() {
submitted++;
futures[i] = threadPool.submit(new Callable<String[]>() {
@Override
public String call() throws Exception {
try {
String origCty = nextLine[columnOrigCty].trim();
float longitude = Float.parseFloat(nextLine[columnLongitude].trim());
float latitude = Float.parseFloat(nextLine[columnLatitude].trim());
public String[] call() throws Exception {
String[] res = new String[inputs.size()];
for (int i = 0; i < inputs.size(); i++) {
String[] nextLine = inputs.get(i);
try {
return countryOfOriginService.getCountries(longitude, latitude, origCty,
ALLOWED_DISTANCE_MARGIN);
} catch (Exception e) {
return e.getMessage();
String origCty = nextLine[columnOrigCty].trim();
float longitude = Float.parseFloat(nextLine[columnLongitude].trim());
float latitude = Float.parseFloat(nextLine[columnLatitude].trim());
try {
res[i] = countryOfOriginService.getCountries(longitude, latitude, origCty,
ALLOWED_DISTANCE_MARGIN);
} catch (Exception e) {
res[i] = e.getMessage();
}
} catch (Throwable e) {
res[i] = "ERROR: " + e.getMessage();
}
} catch (Throwable e) {
return "ERROR: " + e.getMessage();
}
// System.err.println("Processed " + cpuId);
return res;
}
});
}
// Retrieve results
for (int i = nThreads - 1; i >= 0; i--) {
for (int i = 0; i < nThreads; i++) {
// System.err.println("Getting results " + i);
if (futures[i] == null)
break;
String result = null;
final ArrayList<String[]> inputs = (ArrayList<String[]>) inputLines[i];
String[] result = null;
try {
result = (String) futures[i].get();
result = (String[]) futures[i].get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
result = e.getMessage();
}
String[] writeLine = Arrays.copyOf(inputLines[i], inputLines[i].length + 1);
writeLine[writeLine.length - 1] = result;
writer.writeNext(writeLine);
// System.err.println("Results " + i);
for (int j = 0; j < inputs.size(); j++) {
String[] nextLine = inputs.get(j);
String[] writeLine = Arrays.copyOf(nextLine, nextLine.length + 1);
writeLine[writeLine.length - 1] = result[j];
writer.writeNext(writeLine);
}
}
// System.err.println("Submitted " + submitted);
} while (submitted > 0);
threadPool.shutdown();
stopWatch.stop();
writer.close();
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment