Index: contrib/benchmark/src/test/org/apache/lucene/benchmark/byTask/TestPerfTasksLogic.java
===================================================================
--- contrib/benchmark/src/test/org/apache/lucene/benchmark/byTask/TestPerfTasksLogic.java	(revision 575910)
+++ contrib/benchmark/src/test/org/apache/lucene/benchmark/byTask/TestPerfTasksLogic.java	(working copy)
@@ -282,6 +282,39 @@
     assertEquals(totalTokenCount1, totalTokenCount2);
   }
   
+  /**
+   * Test that " {[AddDoc(4000)]: 4} : * " works corrcetly (for LUCENE-941)
+   */
+  public void testParallelExhausted() throws Exception {
+    // 1. alg definition (required in every "logic" test)
+    String algLines[] = {
+        "# ----- properties ",
+        "doc.maker="+Reuters20DocMaker.class.getName(),
+        "doc.add.log.step=3",
+        "doc.term.vector=false",
+        "doc.maker.forever=false",
+        "directory=RAMDirectory",
+        "doc.stored=false",
+        "doc.tokenized=false",
+        "debug.level=1",
+        "# ----- alg ",
+        "CreateIndex",
+        "{ [ AddDoc]: 4} : * ",
+        "ResetInputs ",
+        "{ [ AddDoc]: 4} : * ",
+        "CloseIndex",
+    };
+    
+    // 2. execute the algorithm  (required in every "logic" test)
+    Benchmark benchmark = execBenchmark(algLines);
+
+    // 3. test number of docs in the index
+    IndexReader ir = IndexReader.open(benchmark.getRunData().getDirectory());
+    int ndocsExpected = 2 * 20; // Reuters20DocMaker exhausts after 20 docs.
+    assertEquals("wrong number of docs in the index!", ndocsExpected, ir.numDocs());
+    ir.close();
+  }
+
   // create the benchmark and execute it. 
   public static Benchmark execBenchmark(String[] algLines) throws Exception {
     String algText = algLinesToText(algLines);
@@ -321,5 +354,9 @@
       nDocs++;
       return super.getNextDocData();
     }
+    public synchronized void resetInputs() {
+      super.resetInputs();
+      nDocs = 0;
+    }
   }
 }
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ResetSystemSoftTask.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ResetSystemSoftTask.java	(revision 575910)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ResetSystemSoftTask.java	(working copy)
@@ -28,7 +28,7 @@
  * Index is NOT erased.
  * Directory is NOT erased.
  */
-public class ResetSystemSoftTask extends PerfTask {
+public class ResetSystemSoftTask extends ResetInputsTask {
 
   public ResetSystemSoftTask(PerfRunData runData) {
     super(runData);
@@ -39,12 +39,4 @@
     return 0;
   }
 
-  /*
-   * (non-Javadoc)
-   * @see org.apache.lucene.benchmark.byTask.tasks.PerfTask#shouldNotRecordStats()
-   */
-  protected boolean shouldNotRecordStats() {
-    return true;
-  }
-
 }
Index: contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ResetSystemEraseTask.java
===================================================================
--- contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ResetSystemEraseTask.java	(revision 575910)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/ResetSystemEraseTask.java	(working copy)
@@ -28,7 +28,7 @@
  * Index is erased.
  * Directory is erased.
  */
-public class ResetSystemEraseTask extends PerfTask {
+public class ResetSystemEraseTask extends ResetSystemSoftTask {
 
   public ResetSystemEraseTask(PerfRunData runData) {
     super(runData);
@@ -39,12 +39,4 @@
     return 0;
   }
   
-  /*
-   * (non-Javadoc)
-   * @see org.apache.lucene.benchmark.byTask.tasks.PerfTask#shouldNotRecordStats()
-   */
-  protected boolean shouldNotRecordStats() {
-    return true;
-  }
-
 }
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 575910)
+++ contrib/benchmark/src/java/org/apache/lucene/benchmark/byTask/tasks/TaskSequence.java	(working copy)
@@ -35,7 +35,9 @@
   private boolean letChildReport = true;
   private int rate = 0;
   private boolean perMin = false; // rate, if set, is, by default, be sec.
-  private String seqName; 
+  private String seqName;
+  private boolean exhausted = false;
+  private boolean resetExhausted = false;
   
   public TaskSequence (PerfRunData runData, String name, TaskSequence parent, boolean parallel) {
     super(runData);
@@ -90,6 +92,7 @@
    * @see org.apache.lucene.benchmark.byTask.tasks.PerfTask#doLogic()
    */
   public int doLogic() throws Exception {
+    resetExhausted = false;
     return ( parallel ? doParallelTasks() : doSerialTasks());
   }
 
@@ -99,7 +102,6 @@
     }
     
     int count = 0;
-    boolean exhausted = false;
     
     final int numTasks = tasks.size();
     final PerfTask[] tasksArray = new PerfTask[numTasks];
@@ -110,6 +112,7 @@
       for(int l=0;l<numTasks;l++)
         try {
           count += tasksArray[l].runAndMaybeStats(letChildReport);
+          updateExhausted(tasksArray[l]);
         } catch (NoMoreDataException e) {
           exhausted = true;
         }
@@ -121,7 +124,6 @@
     long delayStep = (perMin ? 60000 : 1000) /rate;
     long nextStartTime = System.currentTimeMillis();
     int count = 0;
-    boolean exhausted = false;
     for (int k=0; (repetitions==REPEAT_EXHAUST && !exhausted) || k<repetitions; k++) {
       for (Iterator it = tasks.iterator(); it.hasNext();) {
         PerfTask task = (PerfTask) it.next();
@@ -133,6 +135,7 @@
         nextStartTime += delayStep; // this aims at avarage rate. 
         try {
           count += task.runAndMaybeStats(letChildReport);
+          updateExhausted(task);
         } catch (NoMoreDataException e) {
           exhausted = true;
         }
@@ -141,6 +144,25 @@
     return count;
   }
 
+  // update state regarding exhaustion.
+  private void updateExhausted(PerfTask task) {
+    if (task instanceof ResetInputsTask) {
+      exhausted = false;
+      resetExhausted = true;
+    } else {
+      if (task instanceof TaskSequence) {
+        TaskSequence t = (TaskSequence) task;
+        if (t.resetExhausted) {
+          exhausted = false;
+          resetExhausted = true;
+          t.resetExhausted = false;
+        } else {
+          exhausted |= t.exhausted;
+        }
+      }
+    }
+  }
+
   private int doParallelTasks() throws Exception {
     final int count [] = {0};
     Thread t[] = new Thread [repetitions * tasks.size()];
@@ -154,12 +176,15 @@
             int n;
             try {
               n = task.runAndMaybeStats(letChildReport);
+              updateExhausted(task);
+              synchronized (count) {
+                count[0] += n;
+              }
+            } catch (NoMoreDataException e) {
+              exhausted = true;
             } catch (Exception e) {
               throw new RuntimeException(e);
             }
-            synchronized (count) {
-              count[0] += n;
-            }
           }
         };
       }