Benchmark custom

lucene/benchmark/src/java/org/apache/lucene/benchmark/byTask/feeds/ReutersContentSource.java

@@ -52,6 +52,8 @@ private static final class DateFormatInfo {
   private ArrayList<Path> inputFiles = new ArrayList<>();
   private int nextFile = 0;
   private int iteration = 0;
+  private int[] threadIndex;
+  private volatile boolean threadIndexCreated;
 
   @Override
   public void setConfig(Config config) {
@@ -102,19 +104,43 @@ public void close() throws IOException {
   public DocData getNextDocData(DocData docData) throws NoMoreDataException, IOException {
     Path f = null;
     String name = null;
-    synchronized (this) {
-      if (nextFile >= inputFiles.size()) {
-        // exhausted files, start a new round, unless forever set to false.
-        if (!forever) {
-          throw new NoMoreDataException();
-        }
-        nextFile = 0;
-        iteration++;
-      }
-      f = inputFiles.get(nextFile++);
-      name = f.toRealPath() + "_" + iteration;
+    int inputFilesSize = inputFiles.size();
+
+    /*
+     * synchronized (this) {
+     * if (nextFile >= inputFiles.size()) { // exhausted files, start a new round, unless forever set to false.
+     * if (!forever) {
+     *    throw new NoMoreDataException();
+     * }
+     * nextFile = 0;
+     * iteration++;
+     * }
+     * f = inputFiles.get(nextFile++);
+     * name = f.toRealPath() + "_" +iteration;
+     * }
+     */
+    if (!threadIndexCreated) {
+      createThreadIndex();
+    }
+
+    int index = (int) Thread.currentThread().getId() % threadIndex.length;
+    int fIndex = index + threadIndex[index] * threadIndex.length;
+    threadIndex[index]++;
+
+    // Sanity check, if # threads is greater than # input files, wrap index
+    if (index >= inputFilesSize) index %= inputFilesSize;
+
+    // Check if this thread has exhausted its files
+    if (fIndex >= inputFilesSize) {
+      threadIndex[index] = 0;
+      fIndex = index + threadIndex[index] * threadIndex.length;
+      threadIndex[index]++;
+      iteration++;
     }
 
+    f = inputFiles.get(fIndex);
+    name = f.toRealPath() + "_" + iteration;
+
     try (BufferedReader reader = Files.newBufferedReader(f, StandardCharsets.UTF_8)) {
       // First line is the date, 3rd is the title, rest is body
       String dateStr = reader.readLine();
@@ -146,4 +172,11 @@ public synchronized void resetInputs() throws IOException {
     nextFile = 0;
     iteration = 0;
   }
+
+  private synchronized void createThreadIndex() {
+    if (!threadIndexCreated) {
+      threadIndex = new int[getConfig().getNumThreads()];
+      threadIndexCreated = true;
+    }
+  }
 }

lucene/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/TaskSequence.java

@@ -334,6 +334,7 @@ private int doParallelTasks() throws Exception {
 
     initTasksArray();
     ParallelTask t[] = runningParallelTasks = new ParallelTask[repetitions * tasks.size()];
+    this.getRunData().getConfig().setNumThreads(t.length);
     // prepare threads
     int index = 0;
     for (int k = 0; k < repetitions; k++) {

lucene/benchmark/src/java/org/apache/lucene/benchmark/byTask/utils/Config.java

@@ -54,6 +54,7 @@ public class Config {
   private HashMap<String, Object> valByRound = new HashMap<>();
   private HashMap<String, String> colForValByRound = new HashMap<>();
   private String algorithmText;
+  private int numThreads = 1;
 
   /**
    * Read both algorithm and config properties.
@@ -113,6 +114,14 @@ public Config(Properties props) {
     }
   }
 
+  public void setNumThreads(int numThreads) {
+    this.numThreads = numThreads;
+  }
+
+  public int getNumThreads() {
+    return numThreads;
+  }
+
   @SuppressWarnings({"unchecked", "rawtypes"})
   private void printProps() {
     System.out.println("------------> config properties:");