Index: contrib/benchmark/CHANGES.txt
===================================================================
--- contrib/benchmark/CHANGES.txt	(revision 881411)
+++ contrib/benchmark/CHANGES.txt	(working copy)
@@ -3,6 +3,11 @@
 The Benchmark contrib package contains code for benchmarking Lucene in a variety of ways.
 
 $Id:$
+11/17/2009
+  LUCENE-2079: Allow specifying delta thread priority after the "&";
+  added log.time.step.msec to print per-time-period counts; fixed
+  NearRealTimeTask to print reopen times (in msec) of each reopen, at
+  the end.  (Mike McCandless)
 
 11/13/2009
   LUCENE-2050: Added ability to run tasks within a serial sequence in
Index: contrib/benchmark/src/test/org/apache/lucene/benchmark/byTask/TestPerfTasksLogic.java
===================================================================
--- contrib/benchmark/src/test/org/apache/lucene/benchmark/byTask/TestPerfTasksLogic.java	(revision 881411)
+++ contrib/benchmark/src/test/org/apache/lucene/benchmark/byTask/TestPerfTasksLogic.java	(working copy)
@@ -123,6 +123,28 @@
     assertTrue("elapsed time was " + elapsed + " msec", elapsed <= 1500);
   }
 
+  public void testBGSearchTaskThreads() throws Exception {
+    String algLines[] = {
+        "log.time.step.msec = 100",
+        "ResetSystemErase",
+        "CreateIndex",
+        "{ AddDoc } : 1000",
+        "Optimize",
+        "CloseIndex",
+        "OpenReader",
+        "{",
+        "  [ \"XSearch\" { CountingSearchTest > : * ] : 2 &-1",
+        "  Wait(1.0)",
+        "}",
+        "CloseReader",
+        "RepSumByPref X"
+    };
+
+    CountingSearchTestTask.numSearches = 0;
+    execBenchmark(algLines);
+    assertTrue(CountingSearchTestTask.numSearches > 0);
+  }
+
   public void testHighlighting() throws Exception {
     // 1. alg definition (required in every "logic" test)
     String algLines[] = {
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/utils/Algorithm.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/utils/Algorithm.java	(revision 881411)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/utils/Algorithm.java	(working copy)
@@ -191,12 +191,22 @@
               if (currSequence.isParallel()) {
                 throw new Exception("Can only create background tasks within a serial task");
               }
+              stok.nextToken();
+              final int deltaPri;
+              if (stok.ttype != StreamTokenizer.TT_NUMBER) {
+                stok.pushBack();
+                deltaPri = 0;
+              } else {
+                // priority
+                deltaPri = (int) stok.nval;
+              }
+
               if (prevTask == null) {
                 throw new Exception("& was unexpected");
               } else if (prevTask.getRunInBackground()) {
                 throw new Exception("double & was unexpected");
               } else {
-                prevTask.setRunInBackground();
+                prevTask.setRunInBackground(deltaPri);
               }
               break;
     
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/NearRealtimeReaderTask.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/NearRealtimeReaderTask.java	(revision 881411)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/NearRealtimeReaderTask.java	(working copy)
@@ -20,6 +20,7 @@
 import org.apache.lucene.benchmark.byTask.PerfRunData;
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.util.ArrayUtil;
 
 /**
  * Spawns a BG thread that periodically (defaults to 3.0
@@ -36,6 +37,9 @@
 
   long pauseMSec = 3000L;
 
+  int reopenCount;
+  int[] reopenTimes = new int[1];
+
   public NearRealtimeReaderTask(PerfRunData runData) {
     super(runData);
   }
@@ -65,7 +69,7 @@
     // stddev, min/max reopen latencies
 
     // Parent sequence sets stopNow
-    int reopenCount = 0;
+    reopenCount = 0;
     while(!stopNow) {
       long waitForMsec = (long) (pauseMSec - (System.currentTimeMillis() - t));
       if (waitForMsec > 0) {
@@ -75,12 +79,16 @@
       t = System.currentTimeMillis();
       final IndexReader newReader = r.reopen();
       if (r != newReader) {
+        final int delay = (int) (System.currentTimeMillis()-t);
+        if (reopenTimes.length == reopenCount) {
+          reopenTimes = ArrayUtil.grow(reopenTimes, 1+reopenCount);
+        }
+        reopenTimes[reopenCount++] = delay;
         // TODO: somehow we need to enable warming, here
         runData.setIndexReader(newReader);
         // Transfer our reference to runData
         newReader.decRef();
         r = newReader;
-        reopenCount++;
       }
     }
 
@@ -94,6 +102,15 @@
   }
 
   @Override
+  public void close() {
+    System.out.println("NRT reopen times:");
+    for(int i=0;i<reopenCount;i++) {
+      System.out.print(" " + reopenTimes[i]);
+    }
+    System.out.println();
+  }
+
+  @Override
   public boolean supportsParams() {
     return true;
   }
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/PerfTask.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/PerfTask.java	(revision 881411)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/PerfTask.java	(working copy)
@@ -61,6 +61,7 @@
   protected String params = null;
 
   private boolean runInBackground;
+  private int deltaPri;
 
   protected static final String NEW_LINE = System.getProperty("line.separator");
 
@@ -72,14 +73,19 @@
     }
   }
 
-  public void setRunInBackground() {
+  public void setRunInBackground(int deltaPri) {
     runInBackground = true;
+    this.deltaPri = deltaPri;
   }
 
   public boolean getRunInBackground() {
     return runInBackground;
   }
 
+  public int getBackgroundDeltaPriority() {
+    return deltaPri;
+  }
+
   protected volatile boolean stopNow;
 
   public void stopNow() {
@@ -216,6 +222,10 @@
     sb.append(getName());
     if (getRunInBackground()) {
       sb.append(" &");
+      int x = getBackgroundDeltaPriority();
+      if (x != 0) {
+        sb.append(x);
+      }
     }
     return sb.toString();
   }
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ReportTask.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ReportTask.java	(revision 881411)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ReportTask.java	(working copy)
@@ -151,11 +151,26 @@
         line = line.replaceAll("   "," - ");
       }
       sb.append(line);
+      int[] byTime = stat.getCountsByTime();
+      if (byTime != null) {
+        sb.append(newline);
+        int end = -1;
+        for(int i=byTime.length-1;i>=0;i--) {
+          if (byTime[i] != 0) {
+            end = i;
+            break;
+          }
+        }
+        if (end != -1) {
+          sb.append("  by time:");
+          for(int i=0;i<end;i++) {
+            sb.append(' ').append(byTime[i]);
+          }
+        }
+      }
     }
+    
     String reptxt = (reported==0 ? "No Matching Entries Were Found!" : sb.toString());
     return new Report(reptxt,partOfTasks.size(),reported,totalSize);
   }
-
-
-
 }
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/TaskSequence.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/TaskSequence.java	(revision 881411)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/TaskSequence.java	(working copy)
@@ -23,6 +23,8 @@
 
 import org.apache.lucene.benchmark.byTask.PerfRunData;
 import org.apache.lucene.benchmark.byTask.feeds.NoMoreDataException;
+import org.apache.lucene.benchmark.byTask.stats.TaskStats;
+import org.apache.lucene.util.ArrayUtil;
 
 /**
  * Sequence of parallel or sequential tasks.
@@ -45,6 +47,7 @@
   
   private boolean fixedTime;                      // true if we run for fixed time
   private double runTimeSec;                      // how long to run for
+  private final long logByTimeMsec;
 
   public TaskSequence (PerfRunData runData, String name, TaskSequence parent, boolean parallel) {
     super(runData);
@@ -55,6 +58,7 @@
     this.parent = parent;
     this.parallel = parallel;
     tasks = new ArrayList<PerfTask>();
+    logByTimeMsec = runData.getConfig().get("log.time.step.msec", 0);
   }
 
   @Override
@@ -76,6 +80,9 @@
         anyExhaustibleTasks |= tasksArray[k] instanceof TaskSequence;
       }
     }
+    if (!parallel && logByTimeMsec != 0 && !letChildReport) {
+      countsByTime = new int[1];
+    }
   }
 
   /**
@@ -92,6 +99,8 @@
     return repetitions;
   }
 
+  private int[] countsByTime;
+
   public void setRunTime(double sec) throws Exception {
     runTimeSec = sec;
     fixedTime = true;
@@ -108,9 +117,6 @@
       if (isParallel()) {
         throw new Exception("REPEAT_EXHAUST is not allowed for parallel tasks");
       }
-      if (getRunData().getConfig().get("content.source.forever",true)) {
-        throw new Exception("REPEAT_EXHAUST requires setting content.source.forever=false");
-      }
     }
     setSequenceName();
   }
@@ -167,11 +173,10 @@
     initTasksArray();
     int count = 0;
 
-    final long t0 = System.currentTimeMillis();
-
     final long runTime = (long) (runTimeSec*1000);
     List<RunBackgroundTask> bgTasks = null;
 
+    final long t0 = System.currentTimeMillis();
     for (int k=0; fixedTime || (repetitions==REPEAT_EXHAUST && !exhausted) || k<repetitions; k++) {
       if (stopNow) {
         break;
@@ -183,11 +188,20 @@
             bgTasks = new ArrayList<RunBackgroundTask>();
           }
           RunBackgroundTask bgTask = new RunBackgroundTask(task, letChildReport);
+          bgTask.setPriority(getBackgroundDeltaPriority() + Thread.currentThread().getPriority());
           bgTask.start();
           bgTasks.add(bgTask);
         } else {
           try {
-            count += task.runAndMaybeStats(letChildReport);
+            final int inc = task.runAndMaybeStats(letChildReport);
+            count += inc;
+            if (countsByTime != null) {
+              final int slot = (int) ((System.currentTimeMillis()-t0)/logByTimeMsec);
+              if (slot >= countsByTime.length) {
+                countsByTime = ArrayUtil.grow(countsByTime, 1+slot);
+              }
+              countsByTime[slot] += inc;
+            }
             if (anyExhaustibleTasks)
               updateExhausted(task);
           } catch (NoMoreDataException e) {
@@ -210,6 +224,11 @@
         count += bgTask.getCount();
       }
     }
+
+    if (countsByTime != null) {
+      getRunData().getPoints().getCurrentStats().setCountsByTime(countsByTime, logByTimeMsec);
+    }
+
     return count;
   }
 
@@ -218,6 +237,7 @@
     long delayStep = (perMin ? 60000 : 1000) /rate;
     long nextStartTime = System.currentTimeMillis();
     int count = 0;
+    final long t0 = System.currentTimeMillis();
     for (int k=0; (repetitions==REPEAT_EXHAUST && !exhausted) || k<repetitions; k++) {
       if (stopNow) {
         break;
@@ -238,7 +258,16 @@
         }
         nextStartTime += delayStep; // this aims at avarage rate. 
         try {
-          count += task.runAndMaybeStats(letChildReport);
+          final int inc = task.runAndMaybeStats(letChildReport);
+          count += inc;
+          if (countsByTime != null) {
+            final int slot = (int) ((System.currentTimeMillis()-t0)/logByTimeMsec);
+            if (slot >= countsByTime.length) {
+              countsByTime = ArrayUtil.grow(countsByTime, 1+slot);
+            }
+            countsByTime[slot] += inc;
+          }
+
           if (anyExhaustibleTasks)
             updateExhausted(task);
         } catch (NoMoreDataException e) {
@@ -305,6 +334,9 @@
   ParallelTask[] runningParallelTasks;
 
   private int doParallelTasks() throws Exception {
+
+    final TaskStats stats = getRunData().getPoints().getCurrentStats();
+
     initTasksArray();
     ParallelTask t[] = runningParallelTasks = new ParallelTask[repetitions * tasks.size()];
     // prepare threads
@@ -323,8 +355,25 @@
     for (int i = 0; i < t.length; i++) {
       t[i].join();
       count += t[i].count;
+      if (t[i].task instanceof TaskSequence) {
+        TaskSequence sub = (TaskSequence) t[i].task;
+        if (sub.countsByTime != null) {
+          if (countsByTime == null) {
+            countsByTime = new int[sub.countsByTime.length];
+          } else if (countsByTime.length < sub.countsByTime.length) {
+            countsByTime = ArrayUtil.grow(countsByTime, sub.countsByTime.length);
+          }
+          for(int j=0;j<sub.countsByTime.length;j++) {
+            countsByTime[j] += sub.countsByTime[j];
+          }
+        }
+      }
     }
 
+    if (countsByTime != null) {
+      stats.setCountsByTime(countsByTime, logByTimeMsec);
+    }
+
     // return total count
     return count;
   }
@@ -386,6 +435,10 @@
     }
     if (getRunInBackground()) {
       sb.append(" &");
+      int x = getBackgroundDeltaPriority();
+      if (x != 0) {
+        sb.append(x);
+      }
     }
     return sb.toString();
   }
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/stats/TaskStats.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/stats/TaskStats.java	(revision 881411)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/stats/TaskStats.java	(working copy)
@@ -91,7 +91,23 @@
     this.numParallelTasks = numParallelTasks;
     this.count = count;
   }
+  
+  private int[] countsByTime;
+  private long countsByTimeStepMSec;
 
+  public void setCountsByTime(int[] counts, long msecStep) {
+    countsByTime = counts;
+    countsByTimeStepMSec = msecStep;
+  }
+
+  public int[] getCountsByTime() {
+    return countsByTime;
+  }
+
+  public long getCountsByTimeStepMSec() {
+    return countsByTimeStepMSec;
+  }
+
   /**
    * @return the taskRunNum.
    */
@@ -174,6 +190,18 @@
     if (round != stat2.round) {
       round = -1; // no meaning if aggregating tasks of different round. 
     }
+
+    if (countsByTime != null && stat2.countsByTime != null) {
+      if (countsByTimeStepMSec != stat2.countsByTimeStepMSec) {
+        throw new IllegalStateException("different by-time msec step");
+      }
+      if (countsByTime.length != stat2.countsByTime.length) {
+        throw new IllegalStateException("different by-time msec count");
+      }
+      for(int i=0;i<stat2.countsByTime.length;i++) {
+        countsByTime[i] += stat2.countsByTime[i];
+      }
+    }
   }
 
   /* (non-Javadoc)
@@ -181,7 +209,11 @@
    */
   @Override
   public Object clone() throws CloneNotSupportedException {
-    return super.clone();
+    TaskStats c = (TaskStats) super.clone();
+    if (c.countsByTime != null) {
+      c.countsByTime = (int[]) c.countsByTime.clone();
+    }
+    return c;
   }
 
   /**
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/stats/Points.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/stats/Points.java	(revision 881411)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/stats/Points.java	(working copy)
@@ -38,6 +38,8 @@
 
   private int nextTaskRunNum = 0;
 
+  private TaskStats currentStats;
+
   /**
    * Create a Points statistics object. 
    */
@@ -62,9 +64,14 @@
    */
   public synchronized TaskStats markTaskStart (PerfTask task, int round) {
     TaskStats stats = new TaskStats(task, nextTaskRunNum(), round);
+    this.currentStats = stats;
     points.add(stats);
     return stats;
   }
+
+  public TaskStats getCurrentStats() {
+    return currentStats;
+  }
   
   // return next task num
   private synchronized int nextTaskRunNum() {