org.gridgain.grid.spi.collision.priorityqueue
Class GridPriorityQueueCollisionSpi

java.lang.Object
  org.gridgain.grid.spi.GridSpiAdapter
      org.gridgain.grid.spi.collision.priorityqueue.GridPriorityQueueCollisionSpi

All Implemented Interfaces:

GridCollisionSpi, GridPriorityQueueCollisionSpiMBean, GridSpi, GridSpiManagementMBean

@GridSpiInfo(author="GridGain Systems",
             url="www.gridgain.org",
             email="support@gridgain.com",
             version="x.x")
@GridSpiMultipleInstancesSupport(value=true)
public class GridPriorityQueueCollisionSpi
extends GridSpiAdapter
implements GridCollisionSpi, GridPriorityQueueCollisionSpiMBean

This class provides implementation for Collision SPI based on priority queue. Jobs are first ordered by their priority, if one is specified, and only first GridPriorityQueueCollisionSpi.getParallelJobsNumber() jobs is allowed to execute in parallel. Other jobs will be queued up.

Configuration

Mandatory

This SPI has no mandatory configuration parameters.

Optional

This SPI has following optional configuration parameters:

• Number of jobs that can be executed in parallel (see GridPriorityQueueCollisionSpi.setParallelJobsNumber(int)). This number should usually be set to no greater than number of threads in the execution thread pool.
• Priority attribute session key (see GridPriorityQueueCollisionSpi.getPriorityAttributeKey()). Prior to returning from GridTask.map(List, Object) method, task implementation should set a value into the task session keyed by this attribute key. See GridTaskSession for more information about task session.
• Default priority value (see GridPriorityQueueCollisionSpi.getDefaultPriority()). It is used when no priority is set.

Below is a Java example of configuration for priority collision SPI:

 GridPriorityQueueCollisionSpi colSpi = new GridPriorityQueueCollisionSpi();

 // Execute all jobs sequentially by setting parallel job number to 1.
 colSpi.setParallelJobsNumber(1);

 GridConfigurationAdapter cfg = new GridConfigurationAdapter();

 // Override default collision SPI.
 cfg.setCollisionSpi(colSpi);

 // Start grid.
 GridFactory.start(cfg);
 

Here is Spring XML configuration example:

 <property name="collisionSpi">
     <bean class="org.gridgain.grid.spi.collision.priorityqueue.GridPriorityQueueCollisionSpi">
         <property name="priorityAttributeKey" value="myPriorityAttributeKey"/>
         <property name="parallelJobsNumber" value="10"/>
     </bean>
 </property>
 

Coding Example

Here is an example of a grid tasks that uses priority collision SPI configured in example above. Note that priority collision resolution is absolutely transparent to the user and is simply a matter of proper grid configuration. Also, priority may be defined only for task (it can be defined within the task, not at a job level). All split jobs will be started with priority declared in their owner task.

This example demonstrates how urgent task may be declared with a higher priority value. Priority SPI guarantees (see its configuration in example above, where number of parallel jobs is set to 1) that all jobs from MyGridUrgentTask will most likely be activated first (one by one) and jobs from MyGridUsualTask with lowest priority will wait. Once higher priority jobs complete, lower priority jobs will be scheduled.

 public class MyGridUsualTask extends GridTaskSplitAdapter<Object, Object> {
    public static final int SPLIT_COUNT = 20;

    @GridTaskSessionResource
    private GridTaskSession taskSes = null;

    @Override
    protected Collection<? extends GridJob> split(int gridSize, Object arg) throws GridException {
        ...
        // Set low task priority (note that attribute name is used by the SPI
        // and should not be changed).
        taskSes.setAttribute("grid.task.priority", 5);

        Collection<GridJob> jobs = new ArrayList<GridJob>(SPLIT_COUNT);

        for (int i = 1; i <= SPLIT_COUNT; i++) {
            jobs.add(new GridJobAdapter<Integer>(i) {
                ...
            });
        }
        ...
    }
 }
 

and

 public class MyGridUrgentTask extends GridTaskSplitAdapter<Object, Object> {
    public static final int SPLIT_COUNT = 5;

    @GridTaskSessionResource
    private GridTaskSession taskSes = null;

    @Override
    protected Collection<? extends GridJob> split(int gridSize, Object arg) throws GridException {
        ...
        // Set high task priority (note that attribute name is used by the SPI
        // and should not be changed).
        taskSes.setAttribute("grid.task.priority", 10);

        Collection<GridJob> jobs = new ArrayList<GridJob>(SPLIT_COUNT);

        for (int i = 1; i <= SPLIT_COUNT; i++) {
            jobs.add(new GridJobAdapter<Integer>(i) {
                ...
            });
        }
        ...
    }
 }
 

For information about Spring framework visit www.springframework.org



See Also:

Documentation
Email Support
Online Forums
Issue Tracking

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

	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

Field Summary

static int	DFLT_PARALLEL_JOBS_NUM Default number of parallel jobs allowed (value is 95 which is slightly less same as default value of threads in the execution thread pool to allow some extra threads for system processing).
static int	DFLT_PRIORITY Default priority that will be assigned if job does not have a priority attribute set (value is 0).
static String	DFLT_PRIORITY_ATTRIBUTE_KEY Default priority attribute key (value is grid.task.priority).

Constructor Summary

GridPriorityQueueCollisionSpi()

Method Summary

protected List<String>	getConsistentAttributeNames() Returns back list of attributes that should be consistent for this SPI.
int	getCurrentActiveJobsNumber() Gets current number of jobs that are being executed.
int	getCurrentWaitJobsNumber() Gets current number of jobs that wait for the execution.
int	getDefaultPriority() Gets default priority to use if a job does not have priority attribute set.
Map<String,Serializable>	getNodeAttributes() This method is called before SPI starts (before method GridSpi.spiStart(String) is called). It allows SPI implementation to add attributes to a local node. Kernel collects these attributes from all SPI implementations loaded up and then passes it to discovery SPI so that they can be exchanged with other nodes.
int	getParallelJobsNumber() Gets number of jobs that can be executed in parallel.
String	getPriorityAttributeKey() Gets key name of priority attribute.
void	onCollision(Collection<GridCollisionJobContext> waitJobs, Collection<GridCollisionJobContext> activeJobs) This is a callback called when either new grid job arrived or executing job finished its execution.
void	setDefaultPriority(int dfltPriority) Sets default job priority.
void	setExternalCollisionListener(GridCollisionExternalListener listener) Listener to be set for notification of external collision events (e.g. job stealing).
void	setParallelJobsNumber(int parallelJobsNum) Sets number of jobs that are allowed to be executed in parallel on this node.
void	setPriorityAttributeKey(String attrKey) Sets task priority attribute key.
void	spiStart(String gridName) This method is called to start SPI.
void	spiStop() This method is called to stop SPI.
String	toString()

Methods inherited from class org.gridgain.grid.spi.GridSpiAdapter

assertParameter, configInfo, createSpiAttributeName, getAuthor, getGridGainHome, getLocalNodeId, getName, getSpiContext, getStartTimestamp, getStartTimestampFormatted, getUpTime, getUpTimeFormatted, getVendorEmail, getVendorUrl, getVersion, onContextDestroyed, onContextInitialized, registerMBean, setName, setSpiContext, startInfo, startStopwatch, stopInfo, unregisterMBean

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.gridgain.grid.spi.GridSpi

getName, onContextDestroyed, onContextInitialized

Methods inherited from interface org.gridgain.grid.spi.GridSpiManagementMBean

getAuthor, getGridGainHome, getLocalNodeId, getName, getStartTimestamp, getStartTimestampFormatted, getUpTime, getUpTimeFormatted, getVendorEmail, getVendorUrl, getVersion

Field Detail

DFLT_PARALLEL_JOBS_NUM

public static final int DFLT_PARALLEL_JOBS_NUM

Default number of parallel jobs allowed (value is 95 which is slightly less same as default value of threads in the execution thread pool to allow some extra threads for system processing).

See Also:

Constant Field Values

DFLT_PRIORITY_ATTRIBUTE_KEY

public static final String DFLT_PRIORITY_ATTRIBUTE_KEY

Default priority attribute key (value is grid.task.priority).

See Also:

Constant Field Values

DFLT_PRIORITY

public static final int DFLT_PRIORITY

Default priority that will be assigned if job does not have a priority attribute set (value is 0).

See Also:

Constant Field Values

Constructor Detail

GridPriorityQueueCollisionSpi

public GridPriorityQueueCollisionSpi()

Method Detail

setParallelJobsNumber

@GridSpiConfiguration(optional=true)
public void setParallelJobsNumber(int parallelJobsNum)

Sets number of jobs that are allowed to be executed in parallel on this node.

If not provided, default value is GridPriorityQueueCollisionSpi.DFLT_PARALLEL_JOBS_NUM.

Parameters:

parallelJobsNum - Maximum number of jobs to be executed in parallel.

getParallelJobsNumber

public int getParallelJobsNumber()

Gets number of jobs that can be executed in parallel.

Specified by:

getParallelJobsNumber in interface GridPriorityQueueCollisionSpiMBean

Returns:

Number of jobs that can be executed in parallel.

getCurrentWaitJobsNumber

public int getCurrentWaitJobsNumber()

Gets current number of jobs that wait for the execution.

Specified by:

getCurrentWaitJobsNumber in interface GridPriorityQueueCollisionSpiMBean

Returns:

Number of jobs that wait for execution.

getCurrentActiveJobsNumber

public int getCurrentActiveJobsNumber()

Gets current number of jobs that are being executed.

Specified by:

getCurrentActiveJobsNumber in interface GridPriorityQueueCollisionSpiMBean

Returns:

Number of active jobs.

setPriorityAttributeKey

@GridSpiConfiguration(optional=true)
public void setPriorityAttributeKey(String attrKey)

Sets task priority attribute key. This key will be used to look up task priorities from task context (see GridTaskSession.getAttribute(Serializable)).

If not provided, default value is GridPriorityQueueCollisionSpi.DFLT_PRIORITY_ATTRIBUTE_KEY.

Parameters:

attrKey - Priority session attribute key.

getPriorityAttributeKey

public String getPriorityAttributeKey()

Gets key name of priority attribute.

Specified by:

getPriorityAttributeKey in interface GridPriorityQueueCollisionSpiMBean

Returns:

Key name of priority attribute.

getDefaultPriority

public int getDefaultPriority()

Gets default priority to use if a job does not have priority attribute set.

Specified by:

getDefaultPriority in interface GridPriorityQueueCollisionSpiMBean

Returns:

Default priority to use if a task does not have priority attribute set.

setDefaultPriority

@GridSpiConfiguration(optional=true)
public void setDefaultPriority(int dfltPriority)

Sets default job priority. If job has no set priority this value will be used to compare with another job.

If not provided, default value is GridPriorityQueueCollisionSpi.DFLT_PRIORITY.

Parameters:

dfltPriority - Default job priority.

getNodeAttributes

public Map<String,Serializable> getNodeAttributes()
                                           throws GridSpiException

This method is called before SPI starts (before method GridSpi.spiStart(String) is called). It allows SPI implementation to add attributes to a local node. Kernel collects these attributes from all SPI implementations loaded up and then passes it to discovery SPI so that they can be exchanged with other nodes.

Specified by:

getNodeAttributes in interface GridSpi

Overrides:

getNodeAttributes in class GridSpiAdapter

Throws:

GridSpiException - Throws in case of any error.

Returns:

Map of local node attributes this SPI wants to add.

spiStart

public void spiStart(String gridName)
              throws GridSpiException

This method is called to start SPI. After this method returns successfully kernel assumes that SPI is fully operational.

Specified by:

spiStart in interface GridSpi

Throws:

GridSpiException - Throws in case of any error during SPI start.

Parameters:

gridName - Name of grid instance this SPI is being started for (null for default grid).

spiStop

public void spiStop()
             throws GridSpiException

This method is called to stop SPI. After this method returns kernel assumes that this SPI is finished and all resources acquired by it are released. Note that this method can be called at any point including during recovery of failed start. It should make no assumptions on what state SPI will be in when this method is called.

Specified by:

spiStop in interface GridSpi

Throws:

GridSpiException - Thrown in case of any error during SPI stop.

setExternalCollisionListener

public void setExternalCollisionListener(GridCollisionExternalListener listener)

Listener to be set for notification of external collision events (e.g. job stealing). Once grid receives such notification, it will immediately invoke collision SPI.

GridGain uses this listener to enable job stealing from overloaded to underloaded nodes. However, you can also utilize it, for instance, to provide time based collision resolution. To achieve this, you most likely would mark some job by setting a certain attribute in job context (see GridJobContext) for a job that requires time-based scheduling and set some timer in your SPI implementation that would wake up after a certain period of time. Once this period is reached, you would notify this listener that a collision resolution should take place. Then inside of your collision resolution logic, you would find the marked waiting job and activate it.

Note that most collision SPI's may not have external collisions. In that case, they should simply ignore this method and do nothing when listener is set.

Specified by:

setExternalCollisionListener in interface GridCollisionSpi

Parameters:

listener - Listener for external collision events.

onCollision

public void onCollision(Collection<GridCollisionJobContext> waitJobs,
                        Collection<GridCollisionJobContext> activeJobs)

This is a callback called when either new grid job arrived or executing job finished its execution. When new job arrives it is added to the end of the wait list and this method is called. When job finished its execution, it is removed from the active list and this method is called (i.e., when grid job is finished it will not appear in any list in collision resolution).

Implementation of this method should act on two lists, each of which contains collision job contexts that define a set of operations available during collision resolution. Refer to GridCollisionJobContext documentation for more information.

Specified by:

onCollision in interface GridCollisionSpi

Parameters:

waitJobs - Ordered collection of collision contexts for jobs that are currently waiting for execution. It can be empty but never null. Note that a new newly arrived job, if any, will always be represented by the last item in this list.

activeJobs - Ordered collection of collision contexts for jobs that are currently executing. It can be empty but never null.

getConsistentAttributeNames

protected List<String> getConsistentAttributeNames()

Returns back list of attributes that should be consistent for this SPI. Consistency means that remote node has to have the same attribute with the same value.

Overrides:

getConsistentAttributeNames in class GridSpiAdapter

Returns:

List or attribute names.

toString

public String toString()

Overrides:

toString in class Object