renamed references to task to slice to be consistent with the underly…

…ing process model

CHANGES.txt

@@ -2,6 +2,8 @@ Cascading Change Log
 
 2.0.0 [unreleased]
 
+  Added the c.m.UnitOfWork interface to give c.f.Flow and c.c.Cascade a common contract.
+
   Changed c.t.h.TupleSerialization#setSerializations() to force TupleSerialization and o.a.h.i.s.WritableSerialization
   are first in the "io.serializations" list.
 

src/core/cascading/flow/FlowProcess.java

@@ -96,13 +96,13 @@ public boolean isCounterStatusInitialized()
       }
 
     @Override
-    public int getNumConcurrentTasks()
+    public int getNumProcessSlices()
       {
       return 0;
       }
 
     @Override
-    public int getCurrentTaskNum()
+    public int getCurrentSliceNum()
       {
       return 0;
       }
@@ -195,9 +195,22 @@ public void setCurrentSession( FlowSession currentSession )
     currentSession.setCurrentProcess( this );
     }
 
-  public abstract int getNumConcurrentTasks();
+  /**
+   * Method getNumProcessSlices returns the number of parallel slices or tasks allocated
+   * for this process execution.
+   * <p/>
+   * For MapReduce platforms, this is the same as the number of tasks for a given MapReduce job.
+   *
+   * @return an int
+   */
+  public abstract int getNumProcessSlices();
 
-  public abstract int getCurrentTaskNum();
+  /**
+   * Method getCurrentSliceNum returns an integer representing which slice instance currently running.
+   *
+   * @return an int
+   */
+  public abstract int getCurrentSliceNum();
 
   /**
    * Method getProperty should be used to return configuration parameters from the underlying system.

src/core/cascading/flow/FlowProcessWrapper.java

@@ -78,15 +78,15 @@ public void setCurrentSession( FlowSession currentSession )
     }
 
   @Override
-  public int getNumConcurrentTasks()
+  public int getNumProcessSlices()
     {
-    return delegate.getNumConcurrentTasks();
+    return delegate.getNumProcessSlices();
     }
 
   @Override
-  public int getCurrentTaskNum()
+  public int getCurrentSliceNum()
     {
-    return delegate.getCurrentTaskNum();
+    return delegate.getCurrentSliceNum();
     }
 
   @Override

src/core/cascading/management/SliceState.java

@@ -34,10 +34,10 @@ public abstract class SliceState extends BaseState
 
   protected SliceState( FlowProcess currentProcess )
     {
-    initialize( currentProcess.getCurrentSession().getCascadingServices(), String.valueOf( currentProcess.getCurrentTaskNum() ) );
+    initialize( currentProcess.getCurrentSession().getCascadingServices(), String.valueOf( currentProcess.getCurrentSliceNum() ) );
     this.sessionID = currentProcess.getCurrentSession().getID();
     this.processID = currentProcess.getID();
-    this.sliceID = String.valueOf( currentProcess.getCurrentTaskNum() );
+    this.sliceID = String.valueOf( currentProcess.getCurrentSliceNum() );
     }
 
   @Override

src/core/cascading/operation/filter/Limit.java

@@ -80,8 +80,8 @@ public void prepare( FlowProcess flowProcess, OperationCall<Context> operationCa
 
     operationCall.setContext( context );
 
-    int numTasks = flowProcess.getNumConcurrentTasks();
-    int taskNum = flowProcess.getCurrentTaskNum();
+    int numTasks = flowProcess.getNumProcessSlices();
+    int taskNum = flowProcess.getCurrentSliceNum();
 
     context.limit = (long) Math.floor( (double) limit / (double) numTasks );
 

src/hadoop/cascading/flow/hadoop/HadoopFlowProcess.java

@@ -157,13 +157,13 @@ public int getCurrentNumReducers()
    * @return int
    */
   @Override
-  public int getCurrentTaskNum()
+  public int getCurrentSliceNum()
     {
     return getJobConf().getInt( "mapred.task.partition", 0 );
     }
 
   @Override
-  public int getNumConcurrentTasks()
+  public int getNumProcessSlices()
     {
     if( isMapper() )
       return getCurrentNumMappers();

src/hadoop/cascading/flow/hadoop/HadoopMapStreamGraph.java

@@ -54,7 +54,7 @@ public HadoopMapStreamGraph( HadoopFlowProcess flowProcess, HadoopFlowStep step,
     setTraps();
     setScopes();
 
-    printGraph( step.getID(), "map", flowProcess.getCurrentTaskNum() );
+    printGraph( step.getID(), "map", flowProcess.getCurrentSliceNum() );
     bind();
     }
 

src/hadoop/cascading/flow/hadoop/HadoopReduceStreamGraph.java

@@ -48,7 +48,7 @@ public HadoopReduceStreamGraph( HadoopFlowProcess flowProcess, HadoopFlowStep st
     setTraps();
     setScopes();
 
-    printGraph( step.getID(), "reduce", flowProcess.getCurrentTaskNum() );
+    printGraph( step.getID(), "reduce", flowProcess.getCurrentSliceNum() );
 
     bind();
     }

src/local/cascading/flow/local/LocalFlowProcess.java

@@ -62,13 +62,13 @@ public void setStepStats( LocalStepStats stepStats )
     }
 
   @Override
-  public int getNumConcurrentTasks()
+  public int getNumProcessSlices()
     {
     return 0;
     }
 
   @Override
-  public int getCurrentTaskNum()
+  public int getCurrentSliceNum()
     {
     return 0;
     }

src/test/cascading/operation/filter/LimitFilterTest.java

@@ -79,13 +79,13 @@ public FlowProcess copyWith( Object object )
       }
 
     @Override
-    public int getNumConcurrentTasks()
+    public int getNumProcessSlices()
       {
       return numTasks;
       }
 
     @Override
-    public int getCurrentTaskNum()
+    public int getCurrentSliceNum()
       {
       return taskNum;
       }