org.gridgain.grid
Interface GridTask<T,R>

Type Parameters:

T - Type of the task argument that is passed into map(List, Object) method.

R - Type of the task result returning from reduce(List) method.

All Superinterfaces:

Serializable

All Known Implementing Classes:

GridifyTaskAdapter, GridifyTaskSplitAdapter, GridTaskAdapter, GridTaskSplitAdapter

@Apache20LicenseCompatible
public interface GridTask<T,R>
extends Serializable

Grid task interface defines a task that can be executed on the grid. Grid task is responsible for splitting business logic into multiple grid jobs, receiving results from individual grid jobs executing on remote nodes, and reducing (aggregating) received jobs' results into final grid task result.

Grid Task Execution Sequence

1. Upon request to execute a grid task with given task name system will find deployed task with given name. Task needs to be deployed prior to execution (see Grid.deployTask(Class) method), however if task does not specify its name explicitly via @GridTaskName annotation, it will be auto-deployed first time it gets executed.
2. System will create new distributed task session (see GridTaskSession).
3. System will inject all annotated resources (including task session) into grid task instance. See org.gridgain.grid.resources package for the list of injectable resources.
4. System will call map(List, Object). This method is responsible for splitting business logic of grid task into multiple grid jobs (units of execution) and mapping them to grid nodes. Method map(List, Object) returns a map of with grid jobs as keys and grid node as values.
5. System will send mapped grid jobs to their respective nodes.
6. Upon arrival on the remote node a grid job will be handled by collision SPI (see GridCollisionSpi) which will determine how a job will be executed on the remote node (immediately, buffered or canceled).
7. Once job execution results become available method result(GridJobResult, List) will be called for each received job result. The policy returned by this method will determine the way task reacts to every job result:
   • If GridJobResultPolicy.WAIT policy is returned, task will continue to wait for other job results. If this result is the last job result, then reduce(List) method will be called.
   • If GridJobResultPolicy.REDUCE policy is returned, then method reduce(List) will be called right away without waiting for other jobs' completion (all remaining jobs will receive a cancel request).
   • If GridJobResultPolicy.FAILOVER policy is returned, then job will be failed over to another node for execution. The node to which job will get failed over is decided by GridFailoverSpi SPI implementation. Note that if you use GridTaskAdapter adapter for GridTask implementation, then it will automatically fail jobs to another node for 2 known failure cases:
     • Job has failed due to node crash. In this case GridJobResult.getException() method will return an instance of GridTopologyException exception.
     • Job execution was rejected, i.e. remote node has cancelled job before it got a chance to execute, while it still was on the waiting list. In this case GridJobResult.getException() method will return an instance of GridExecutionRejectedException exception.
8. Once all results are received or result(GridJobResult, List) method returned GridJobResultPolicy.REDUCE policy, method reduce(List) is called to aggregate received results into one final result. Once this method is finished the execution of the grid task is complete. This result will be returned to the user through GridTaskFuture.get() method.

Resource Injection

Grid task implementation can be injected using IoC (dependency injection) with grid resources. Both, field and method based injection are supported. The following grid resources can be injected:

• GridTaskSessionResource
• GridInstanceResource
• GridLoggerResource
• GridHomeResource
• GridExecutorServiceResource
• GridLocalNodeIdResource
• GridMBeanServerResource

Refer to corresponding resource documentation for more information.

Grid Task Adapters

GridTask comes with several convenience adapters to make the usage easier:

• GridTaskAdapter provides default implementation for GridTask.result(GridJobResult, List) method which provides automatic fail-over to another node if remote job has failed due to node crash (detected by GridTopologyException exception) or due to job execution rejection (detected by GridExecutionRejectedException exception). Here is an example of how a you would implement your task using GridTaskAdapter:

   public class MyFooBarTask extends GridTaskAdapter<String, String> {
       // Inject load balancer.
       @GridLoadBalancerResource
       GridLoadBalancer balancer;
   
       // Map jobs to grid nodes.
       public Map<? extends GridJob, GridNode> map(List<GridNode> subgrid, String arg) throws GridException {
           Map<MyFooBarJob, GridNode> jobs = new HashMap<MyFooBarJob, GridNode>(subgrid.size());
   
           // In more complex cases, you can actually do
           // more complicated assignments of jobs to nodes.
           for (int i = 0; i < subgrid.size(); i++) {
               // Pick the next best balanced node for the job.
               jobs.put(new MyFooBarJob(arg), balancer.getBalancedNode())
           }
   
           return jobs;
       }
   
       // Aggregate results into one compound result.
       public String reduce(List<GridJobResult> results) throws GridException {
           // For the purpose of this example we simply
           // concatenate string representation of every 
           // job result
           StringBuilder buf = new StringBuilder();
   
           for (GridJobResult res : results) {
               // Append string representation of result
               // returned by every job.
               buf.append(res.getData().toString());
           }
   
           return buf.toString();
       }
   } 
   

• GridTaskSplitAdapter hides the job-to-node mapping logic from user and provides convenient GridTaskSplitAdapter.split(int, Object) method for splitting task into sub-jobs in homogeneous environments. Here is an example of how you would implement your task using GridTaskSplitAdapter:

   public class MyFooBarTask extends GridTaskSplitAdapter<Object, String> {
       @Override
       protected Collection<? extends GridJob> split(int gridSize, Object arg) throws GridException {
           List<MyFooBarJob> jobs = new ArrayList<MyFooBarJob>(gridSize);
   
           for (int i = 0; i < gridSize; i++) {
               jobs.add(new MyFooBarJob(arg));
           }
   
           // Node assignment via load balancer 
           // happens automatically.
           return jobs;
       }
   
       // Aggregate results into one compound result.
       public String reduce(List<GridJobResult> results) throws GridException {
           // For the purpose of this example we simply
           // concatenate string representation of every 
           // job result
           StringBuilder buf = new StringBuilder();
   
           for (GridJobResult res : results) {
               // Append string representation of result
               // returned by every job.
               buf.append(res.getData().toString());
           }
   
           return buf.toString();
       }
   } 
   

Refer to corresponding adapter documentation for more information.

Map/Reduce

The design of GridTask is influenced by Google Map/Reduce paradigm. For more information about Map/Reduce paradigm, refer to Map/Reduce: Simplified Data Processing on Large Clusters article from Google. Note, however, that in Google terminology task means GridGain's job and job means GridGain's task. Following diagram shows basic idea behind Map/Reduce:

1. Task execution request
2. Task splits (maps) into jobs
3. Result of job execution
4. Aggregation (reduction) of job results into task result

Examples

Many task example usages are available on GridGain Wiki. To see example on how to use GridTask for basic split/aggregate logic refer to HelloWorld Task Example. For example on how to use GridTask with automatic grid-enabling via @Gridify annotation refer to Gridify HelloWorld Example.

Migrating to GridGain 2.0

In GridGain 2.0 this interface API has been updated for better static type checking. Although the change is trivial and provides much better type safety during development - it introduced incompatibility with prior versions of GridGain. Follow this link for easy source code migration instructions.



See Also:

Documentation
Email Support
Online Forums
Issue Tracking

Author: 2005-2008 Copyright © GridGain Systems. All Rights Reserved. ver. 2.0.3

	Nikita Ivanov, GridGain Systems nivanov@gridgain.com
	Dmitriy Setrakyan, GridGain Systems dsetrakyan@gridgain.com
	Alexander Magdenko, GridGain Systems amagdenko@gridgain.com
	Boris Suchodoev, GridGain Systems bsuchodoev@gridgain.com
	Denis Kharlamov, GridGain Systems dkharlamov@gridgain.com
	close

Method Summary

Map<? extends GridJob,GridNode>	map(List<GridNode> subgrid, T arg) This method is called to map or split grid task into multiple grid jobs.
R	reduce(List<GridJobResult> results) Reduces (or aggregates) results received so far into one compound result to be returned to caller via GridTaskFuture.get() method.
GridJobResultPolicy	result(GridJobResult result, List<GridJobResult> received) Asynchronous callback invoked every time a result from remote execution is received.

Method Detail

map

Map<? extends GridJob,GridNode> map(List<GridNode> subgrid,
                                    @Nullable
                                    T arg)
                                    throws GridException

This method is called to map or split grid task into multiple grid jobs. This is the first method that gets called when task execution starts.

Throws:

GridException - If mapping could not complete successfully. This exception will be thrown out of GridTaskFuture.get() method.

Parameters:

arg - Task execution argument. Can be null. This is the same argument as the one passed into Grid.execute(GridTask, Object) method (or any of the other Grid.execute(...) methods.

subgrid - Nodes available for this task execution. Note that order of nodes is guaranteed to be randomized by container. This ensures that every time you simply iterate through grid nodes, the order of nodes will be random which over time should result into all nodes being used equally.

Returns:

Map of grid jobs assigned to subgrid node. If null or empty list is returned, exception will be thrown.

result

GridJobResultPolicy result(GridJobResult result,
                           List<GridJobResult> received)
                           throws GridException

Asynchronous callback invoked every time a result from remote execution is received. It is ultimately upto this method to return a policy based on which the system will either wait for more results, reduce results received so far, or failover this job to another node. See GridJobResultPolicy for more information about result policies.

Throws:

GridException - If handling a job result caused an error. This exception will be thrown out of GridTaskFuture.get() method.

Parameters:

result - Received remote grid executable result.

received - All previously received results. Note that if task class has GridTaskNoResultCache annotation, actual job result won't be cached and GridJobResult.getData() method will return null.

Returns:

Result policy that dictates how to process further upcoming job results.

reduce

R reduce(List<GridJobResult> results)
         throws GridException

Reduces (or aggregates) results received so far into one compound result to be returned to caller via GridTaskFuture.get() method.

Note, that if some jobs did not succeed and could not be failed over then the list of results passed into this method will include the failed results. Otherwise, failed results will not be in the list.

Throws:

GridException - If reduction or results caused an error. This exception will be thrown out of GridTaskFuture.get() method.

Parameters:

results - Received results of broadcasted remote executions. Note that if task class has GridTaskNoResultCache annotation, actual job results won't be cached and GridJobResult.getData() methods will return null.

Returns:

Grid job result constructed from results of remote executions.