Package org.apache.ignite.spi.collision.priorityqueue

## Class PriorityQueueCollisionSpi

- java.lang.Object
- - org.apache.ignite.spi.IgniteSpiAdapter
  - - org.apache.ignite.spi.collision.priorityqueue.PriorityQueueCollisionSpi

- All Implemented Interfaces:
    `CollisionSpi`, `IgniteSpi`

      @IgniteSpiMultipleInstancesSupport(true)
      @IgniteSpiConsistencyChecked(optional=true)
      public class PriorityQueueCollisionSpi
      extends IgniteSpiAdapter
      implements CollisionSpi

  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 `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 `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 `getPriorityAttributeKey()`). Prior to returning from `ComputeTask.map(List, Object)` method, task implementation should set a value into the task session keyed by this attribute key. See `ComputeTaskSession` for more information about task session.
  - Priority attribute job context key (see `getJobPriorityAttributeKey()`). It is used for specifying job priority. See `ComputeJobContext` for more information about job context.
  - Default priority value (see `getDefaultPriority()`). It is used when no priority is set.
  - Default priority increase value (see `getStarvationIncrement()`). It is used for increasing priority when job gets bumped down. This future is used for preventing starvation waiting jobs execution.
  - Default increasing priority flag value (see `isStarvationPreventionEnabled()`). It is used for enabling increasing priority when job gets bumped down. This future is used for preventing starvation waiting jobs execution.
  Below is a Java example of configuration for priority collision SPI:

       PriorityQueueCollisionSpi colSpi = new PriorityQueueCollisionSpi();

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

       IgniteConfiguration cfg = new IgniteConfiguration();

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

       // Starts grid.
       G.start(cfg);

  Here is Spring XML configuration example:

       <property name="collisionSpi">
           <bean class="org.apache.ignite.spi.collision.priorityqueue.PriorityQueueCollisionSpi">
               <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 ComputeTaskSplitAdapter<Object, Object> {
          public static final int SPLIT_COUNT = 20;

          @TaskSessionResource
          private ComputeTaskSession taskSes;

          @Override
          protected Collection<? extends ComputeJob> split(int gridSize, Object arg) throws IgniteCheckedException {
              ...
              // 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<ComputeJob> jobs = new ArrayList<ComputeJob>(SPLIT_COUNT);

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

  and

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

          @TaskSessionResource
          private ComputeTaskSession taskSes;

          @Override
          protected Collection<? extends ComputeJob> split(int gridSize, Object arg) throws IgniteCheckedException {
              ...
              // 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<ComputeJob> jobs = new ArrayList<ComputeJob>(SPLIT_COUNT);

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

  For information about Spring framework visit www.springframework.org

- - ### Field Summary

    Fields
    Modifier and Type | Field | Description
    --- | --- | ---
    `static String` | `DFLT_JOB_PRIORITY_ATTRIBUTE_KEY` | Default job priority attribute key (value is `grid.job.priority`).
    `static int` | `DFLT_PARALLEL_JOBS_NUM` | Default number of parallel jobs allowed (set to number of cores times 2).
    `static boolean` | `DFLT_PREVENT_STARVATION_ENABLED` | Default flag for preventing starvation of lower priority jobs.
    `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`).
    `static int` | `DFLT_STARVATION_INCREMENT` | Default value on which job priority will be increased every time when job gets bumped down.
    `static int` | `DFLT_WAIT_JOBS_NUM` | Default waiting jobs number.

    - ### Fields inherited from class org.apache.ignite.spi.IgniteSpiAdapter

      `ignite, igniteInstanceName`

  - ### Constructor Summary

    Constructors
    Constructor | Description
    --- | ---
    `PriorityQueueCollisionSpi()` |

  - ### Method Summary

    All Methods Instance Methods Concrete Methods
    Modifier and Type | Method | Description
    --- | --- | ---
    `protected List<String>` | `getConsistentAttributeNames()` | Returns back a list of attributes that should be consistent for this SPI.
    `int` | `getCurrentActiveJobsNumber()` | Gets current number of jobs that are active, i.e.
    `int` | `getCurrentHeldJobsNumber()` | Gets number of currently `'held'` jobs.
    `int` | `getCurrentRunningJobsNumber()` | Gets number of currently running (not `'held`) jobs.
    `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.
    `String` | `getJobPriorityAttributeKey()` | Gets key name of job priority attribute.
    `Map<String,​Object>` | `getNodeAttributes()` | This method is called before SPI starts (before method `IgniteSpi.spiStart(String)` is called).
    `int` | `getParallelJobsNumber()` | Gets number of jobs that can be executed in parallel.
    `String` | `getPriorityAttributeKey()` | Gets key name of task priority attribute.
    `int` | `getStarvationIncrement()` | Gets value to increment job priority by every time a lower priority job gets behind a higher priority job.
    `int` | `getWaitingJobsNumber()` | Maximum number of jobs that are allowed to wait in waiting queue.
    `boolean` | `isStarvationPreventionEnabled()` | Gets flag indicating whether job starvation prevention is enabled.
    `void` | `onCollision​(CollisionContext ctx)` | This is a callback called: new grid job arrived executing job finished its execution topology changed periodically (on `EventType.EVT_NODE_METRICS_UPDATED`) When new job arrives it is added to the end of the wait list and this method is called.
    `PriorityQueueCollisionSpi` | `setDefaultPriority​(int priority)` | Sets default priority to use if a job does not have priority attribute set.
    `void` | `setExternalCollisionListener​(CollisionExternalListener lsnr)` | Listener to be set for notification of external collision events (e.g. job stealing).
    `PriorityQueueCollisionSpi` | `setJobPriorityAttributeKey​(String jobPriAttrKey)` | Sets job priority attribute key.
    `PriorityQueueCollisionSpi` | `setName​(String name)` | Sets SPI name.
    `PriorityQueueCollisionSpi` | `setParallelJobsNumber​(int parallelJobsNum)` | Sets number of jobs that can be executed in parallel.
    `PriorityQueueCollisionSpi` | `setPriorityAttributeKey​(String taskPriAttrKey)` | Sets task priority attribute key.
    `PriorityQueueCollisionSpi` | `setStarvationIncrement​(int starvationInc)` | Sets value to increment job priority by every time a lower priority job gets behind a higher priority job.
    `PriorityQueueCollisionSpi` | `setStarvationPreventionEnabled​(boolean preventStarvation)` | Sets flag indicating whether job starvation prevention is enabled.
    `PriorityQueueCollisionSpi` | `setWaitingJobsNumber​(int waitJobsNum)` | Maximum number of jobs that are allowed to wait in waiting queue.
    `void` | `spiStart​(String igniteInstanceName)` | This method is called to start SPI.
    `void` | `spiStop()` | This method is called to stop SPI.
    `String` | `toString()`

    - ### Methods inherited from class org.apache.ignite.spi.IgniteSpiAdapter

      `addTimeoutObject, assertParameter, checkConfigurationConsistency0, clientFailureDetectionTimeout, configInfo, createSpiAttributeName, failureDetectionTimeout, failureDetectionTimeoutEnabled, failureDetectionTimeoutEnabled, getExceptionRegistry, getLocalNode, getName, getSpiContext, ignite, initFailureDetectionTimeout, injectables, injectResources, isNodeStopping, onBeforeStart, onClientDisconnected, onClientReconnected, onContextDestroyed, onContextDestroyed0, onContextInitialized, onContextInitialized0, registerMBean, removeTimeoutObject, started, 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.apache.ignite.spi.IgniteSpi

      `getName, onClientDisconnected, onClientReconnected, onContextDestroyed, onContextInitialized`

- - ### Field Detail

    - #### DFLT_PARALLEL_JOBS_NUM

          public static final int DFLT_PARALLEL_JOBS_NUM

      Default number of parallel jobs allowed (set to number of cores times 2).

    - #### DFLT_WAIT_JOBS_NUM

          public static final int DFLT_WAIT_JOBS_NUM

      Default waiting jobs number. If number of waiting jobs exceed this number, jobs will be rejected. Default value is `Integer.MAX_VALUE`.
      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_JOB_PRIORITY_ATTRIBUTE_KEY

          public static final String DFLT_JOB_PRIORITY_ATTRIBUTE_KEY

      Default job priority attribute key (value is `grid.job.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

    - #### DFLT_STARVATION_INCREMENT

          public static final int DFLT_STARVATION_INCREMENT

      Default value on which job priority will be increased every time when job gets bumped down.
      See Also:
        Constant Field Values

    - #### DFLT_PREVENT_STARVATION_ENABLED

          public static final boolean DFLT_PREVENT_STARVATION_ENABLED

      Default flag for preventing starvation of lower priority jobs.
      See Also:
        Constant Field Values

  - ### Constructor Detail

    - #### PriorityQueueCollisionSpi

          public PriorityQueueCollisionSpi()

  - ### Method Detail

    - #### getParallelJobsNumber

          public int getParallelJobsNumber()

      Gets number of jobs that can be executed in parallel.
      Returns:
        Number of jobs that can be executed in parallel.

    - #### setParallelJobsNumber

          @IgniteSpiConfiguration(optional=true)
          public PriorityQueueCollisionSpi setParallelJobsNumber​(int parallelJobsNum)

      Sets number of jobs that can be executed in parallel.
      Parameters:
        `parallelJobsNum` - Parallel jobs number.
      Returns:
        `this` for chaining.

    - #### getWaitingJobsNumber

          public int getWaitingJobsNumber()

      Maximum number of jobs that are allowed to wait in waiting queue. If number of waiting jobs ever exceeds this number, excessive jobs will be rejected.
      Returns:
        Maximum allowed number of waiting jobs.

    - #### setWaitingJobsNumber

          @IgniteSpiConfiguration(optional=true)
          public PriorityQueueCollisionSpi setWaitingJobsNumber​(int waitJobsNum)

      Maximum number of jobs that are allowed to wait in waiting queue. If number of waiting jobs ever exceeds this number, excessive jobs will be rejected.
      Parameters:
        `waitJobsNum` - Maximium jobs number.
      Returns:
        `this` for chaining.

    - #### getCurrentWaitJobsNumber

          public int getCurrentWaitJobsNumber()

      Gets current number of jobs that wait for the execution.
      Returns:
        Number of jobs that wait for execution.

    - #### getCurrentActiveJobsNumber

          public int getCurrentActiveJobsNumber()

      Gets current number of jobs that are active, i.e. `'running + held'` jobs.
      Returns:
        Number of active jobs.

    - #### getCurrentRunningJobsNumber

          public int getCurrentRunningJobsNumber()

      Gets number of currently running (not `'held`) jobs.
      Returns:
        Number of currently running (not `'held`) jobs.

    - #### getCurrentHeldJobsNumber

          public int getCurrentHeldJobsNumber()

      Gets number of currently `'held'` jobs.
      Returns:
        Number of currently `'held'` jobs.

    - #### setPriorityAttributeKey

          @IgniteSpiConfiguration(optional=true)
          public PriorityQueueCollisionSpi setPriorityAttributeKey​(String taskPriAttrKey)

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

      If not provided, default value is `{@link #DFLT_PRIORITY_ATTRIBUTE_KEY}`.

      Parameters:
        `taskPriAttrKey` - Priority session attribute key.
      Returns:
        `this` for chaining.

    - #### setJobPriorityAttributeKey

          @IgniteSpiConfiguration(optional=true)
          public PriorityQueueCollisionSpi setJobPriorityAttributeKey​(String jobPriAttrKey)

      Sets job priority attribute key. This key will be used to look up job priorities from job context (see `ComputeJobContext.getAttribute(Object)`).

      If not provided, default value is `{@link #DFLT_JOB_PRIORITY_ATTRIBUTE_KEY}`.

      Parameters:
        `jobPriAttrKey` - Job priority attribute key.
      Returns:
        `this` for chaining.

    - #### getPriorityAttributeKey

          public String getPriorityAttributeKey()

      Gets key name of task priority attribute.
      Returns:
        Key name of task priority attribute.

    - #### getJobPriorityAttributeKey

          public String getJobPriorityAttributeKey()

      Gets key name of job priority attribute.
      Returns:
        Key name of job priority attribute.

    - #### getDefaultPriority

          public int getDefaultPriority()

      Gets default priority to use if a job does not have priority attribute set.
      Returns:
        Default priority to use if a task does not have priority attribute set.

    - #### setDefaultPriority

          @IgniteSpiConfiguration(optional=true)
          public PriorityQueueCollisionSpi setDefaultPriority​(int priority)

      Sets default priority to use if a job does not have priority attribute set.
      Parameters:
        `priority` - default priority.
      Returns:
        `this` for chaining.

    - #### getStarvationIncrement

          public int getStarvationIncrement()

      Gets value to increment job priority by every time a lower priority job gets behind a higher priority job.
      Returns:
        Value to increment job priority by every time a lower priority job gets behind a higher priority job.

    - #### setStarvationIncrement

          @IgniteSpiConfiguration(optional=true)
          public PriorityQueueCollisionSpi setStarvationIncrement​(int starvationInc)

      Sets value to increment job priority by every time a lower priority job gets behind a higher priority job.
      Parameters:
        `starvationInc` - Increment value.
      Returns:
        `this` for chaining.

    - #### isStarvationPreventionEnabled

          public boolean isStarvationPreventionEnabled()

      Gets flag indicating whether job starvation prevention is enabled.
      Returns:
        Flag indicating whether job starvation prevention is enabled.

    - #### setStarvationPreventionEnabled

          @IgniteSpiConfiguration(optional=true)
          public PriorityQueueCollisionSpi setStarvationPreventionEnabled​(boolean preventStarvation)

      Sets flag indicating whether job starvation prevention is enabled.
      Parameters:
        `preventStarvation` - Flag indicating whether job starvation prevention is enabled.
      Returns:
        `this` for chaining.

    - #### getNodeAttributes

          public Map<String,​Object> getNodeAttributes()
                                                     throws IgniteSpiException

      This method is called before SPI starts (before method `IgniteSpi.spiStart(String)` is called). It allows SPI implementation to add attributes to a local node. Kernal 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 `IgniteSpi`
      Overrides:
        `getNodeAttributes` in class `IgniteSpiAdapter`
      Returns:
        Map of local node attributes this SPI wants to add.
      Throws:
        `IgniteSpiException` - Throws in case of any error.

    - #### spiStart

          public void spiStart​(String igniteInstanceName)
                        throws IgniteSpiException

      This method is called to start SPI. After this method returns successfully kernel assumes that SPI is fully operational.
      Specified by:
        `spiStart` in interface `IgniteSpi`
      Parameters:
        `igniteInstanceName` - Name of Ignite instance this SPI is being started for (`null` for default Ignite instance).
      Throws:
        `IgniteSpiException` - Throws in case of any error during SPI start.

    - #### spiStop

          public void spiStop()
                       throws IgniteSpiException

      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 `IgniteSpi`
      Throws:
        `IgniteSpiException` - Thrown in case of any error during SPI stop.

    - #### setExternalCollisionListener

          public void setExternalCollisionListener​(CollisionExternalListener lsnr)

      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.

      Ignite 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 `ComputeJobContext`) 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 might 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 `CollisionSpi`
      Parameters:
        `lsnr` - Listener for external collision events.

    - #### onCollision

          public void onCollision​(CollisionContext ctx)

      This is a callback called:
      - new grid job arrived
      - executing job finished its execution
      - topology changed
      - periodically (on `EventType.EVT_NODE_METRICS_UPDATED`)
      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 all lists, each of which contains collision job contexts that define a set of operations available during collision resolution. Refer to `CollisionContext` and `CollisionJobContext` documentation for more information.

      Specified by:
        `onCollision` in interface `CollisionSpi`
      Parameters:
        `ctx` - Collision context which contains all collision lists.

    - #### getConsistentAttributeNames

          protected List<String> getConsistentAttributeNames()

      Returns back a 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 `IgniteSpiAdapter`
      Returns:
        List or attribute names.

    - #### setName

          public PriorityQueueCollisionSpi setName​(String name)

      Sets SPI name.
      Overrides:
        `setName` in class `IgniteSpiAdapter`
      Parameters:
        `name` - SPI name.
      Returns:
        `this` for chaining.

    - #### toString

          public String toString()

      Overrides:
        `toString` in class `Object`