Package org.apache.ignite.compute

## Interface ComputeTaskSession

-     public interface ComputeTaskSession

  Defines a distributed session for particular task execution.

  # Description

  This interface defines a distributed session that exists for particular task execution. Task session is distributed across the parent task and all grid jobs spawned by it, so attributes set on a task or on a job can be viewed on other jobs. Correspondingly attributes set on any of the jobs can also be viewed on a task.

  Session has 2 main features: `attribute` and `checkpoint` management. Both attributes and checkpoints can be used from task itself and from the jobs belonging to this task. Session attributes and checkpoints can be set from any task or job methods. Session attribute and checkpoint consistency is fault tolerant and is preserved whenever a job gets failed over to another node for execution. Whenever task execution ends, all checkpoints saved within session with `ComputeTaskSessionScope.SESSION_SCOPE` scope will be removed from checkpoint storage. Checkpoints saved with `ComputeTaskSessionScope.GLOBAL_SCOPE` will outlive the session and can be viewed by other tasks.

  The sequence in which session attributes are set is consistent across the task and all job siblings within it. There will never be a case when one job sees attribute A before attribute B, and another job sees attribute B before A. Attribute order is identical across all session participants. Attribute order is also fault tolerant and is preserved whenever a job gets failed over to another node.

  # Connected Tasks

  Note that apart from setting and getting session attributes, tasks or jobs can choose to wait for a certain attribute to be set using any of the `waitForAttribute(...)` methods. Tasks and jobs can also receive asynchronous notifications about a certain attribute being set through `ComputeTaskSessionAttributeListener` listener. Such feature allows grid jobs and tasks remain connected in order to synchronize their execution with each other and opens a solution for a whole new range of problems.

  Imagine for example that you need to compress a very large file (let's say terabytes in size). To do that in grid environment you would split such file into multiple sections and assign every section to a remote job for execution. Every job would have to scan its section to look for repetition patterns. Once this scan is done by all jobs in parallel, jobs would need to synchronize their results with their siblings so compression would happen consistently across the whole file. This can be achieved by setting repetition patterns discovered by every job into the session. Once all patterns are synchronized, all jobs can proceed with compressing their designated file sections in parallel, taking into account repetition patterns found by all the jobs in the split. Grid task would then reduce (aggregate) all compressed sections into one compressed file. Without session attribute synchronization step this problem would be much harder to solve.

  # Session Injection

  Session can be injected into a task or a job using IoC (dependency injection) by attaching `@TaskSessionResource` annotation to a field or a setter method inside of `ComputeTask` or `ComputeJob` implementations as follows:

       ...
       // This field will be injected with distributed task session.
       @TaskSessionResource
       private ComputeTaskSession ses;
       ...

  or from a setter method:

       // This setter method will be automatically called by the system
       // to set grid task session.
       @TaskSessionResource
       void setSession(ComputeTaskSession ses) {
           this.ses = ses;
       }

  # Example

- - ### Method Summary

    All Methods Instance Methods Abstract Methods
    Modifier and Type | Method | Description
    --- | --- | ---
    `void` | `addAttributeListener​(ComputeTaskSessionAttributeListener lsnr, boolean rewind)` | Add listener for the session attributes.
    `<K,​V> V` | `getAttribute​(K key)` | Gets an attribute set by `setAttribute(Object, Object)` or `setAttributes(Map)` method.
    `Map<?,​?>` | `getAttributes()` | Gets all attributes.
    `ClassLoader` | `getClassLoader()` | Gets class loader responsible for loading all classes within task.
    `long` | `getEndTime()` | Gets end of computation time for the task.
    `IgniteUuid` | `getId()` | Gets session ID of the task being executed.
    `@Nullable ComputeJobSibling` | `getJobSibling​(IgniteUuid jobId)` | Gets job sibling for a given ID.
    `Collection<ComputeJobSibling>` | `getJobSiblings()` | Gets a collection of all grid job siblings.
    `long` | `getStartTime()` | Gets start of computation time for the task.
    `String` | `getTaskName()` | Gets task name of the task this session belongs to.
    `UUID` | `getTaskNodeId()` | Gets ID of the node on which task execution originated.
    `Collection<UUID>` | `getTopology()` | Gets a collection of grid nodes IDs.
    `<T> T` | `loadCheckpoint​(String key)` | Loads job's state previously saved via `saveCheckpoint(String, Object, ComputeTaskSessionScope, long)` method from an underlying storage for a given `key`.
    `IgniteFuture<?>` | `mapFuture()` | Gets future that will be completed when task "`map`" step has completed (which means that `ComputeTask.map(List, Object)` method has finished).
    `Collection<ComputeJobSibling>` | `refreshJobSiblings()` | Refreshes collection of job siblings.
    `boolean` | `removeAttributeListener​(ComputeTaskSessionAttributeListener lsnr)` | Removes given listener.
    `boolean` | `removeCheckpoint​(String key)` | Removes previously saved job's state for a given `key` from an underlying storage.
    `void` | `saveCheckpoint​(String key, Object state)` | Saves intermediate state of a job or task to a storage.
    `void` | `saveCheckpoint​(String key, Object state, ComputeTaskSessionScope scope, long timeout)` | Saves intermediate state of a job to a storage.
    `void` | `saveCheckpoint​(String key, Object state, ComputeTaskSessionScope scope, long timeout, boolean overwrite)` | Saves intermediate state of a job or task to a storage.
    `void` | `setAttribute​(Object key, @Nullable Object val)` | Sets session attributed.
    `void` | `setAttributes​(Map<?,​?> attrs)` | Sets task attributes.
    `<K,​V> V` | `waitForAttribute​(K key, long timeout)` | Waits for the specified attribute to be set.
    `<K,​V> boolean` | `waitForAttribute​(K key, V val, long timeout)` | Waits for the specified attribute to be set or updated with given value.
    `Map<?,​?>` | `waitForAttributes​(Collection<?> keys, long timeout)` | Waits for the specified attributes to be set.
    `boolean` | `waitForAttributes​(Map<?,​?> attrs, long timeout)` | Waits for the specified attributes to be set or updated with given values.

- - ### Method Detail

    - #### getTaskName

          String getTaskName()

      Gets task name of the task this session belongs to.
      Returns:
        Task name of the task this session belongs to.

    - #### getTaskNodeId

          UUID getTaskNodeId()

      Gets ID of the node on which task execution originated.
      Returns:
        ID of the node on which task execution originated.

    - #### getStartTime

          long getStartTime()

      Gets start of computation time for the task.
      Returns:
        Start of computation time for the task.

    - #### getEndTime

          long getEndTime()

      Gets end of computation time for the task. No job within the task will be allowed to execute passed this time.
      Returns:
        End of computation time for the task.

    - #### getId

          IgniteUuid getId()

      Gets session ID of the task being executed.
      Returns:
        Session ID of the task being executed.

    - #### getClassLoader

          ClassLoader getClassLoader()

      Gets class loader responsible for loading all classes within task.

      Note that for classes that were loaded remotely from other nodes methods `Class.getResource(String)` or `ClassLoader.getResource(String)` will always return `null`. Use `Class.getResourceAsStream(String)` or `ClassLoader.getResourceAsStream(String)` instead.

      Returns:
        Class loader responsible for loading all classes within task.

    - #### getJobSiblings

          Collection<ComputeJobSibling> getJobSiblings()
                                                throws IgniteException

      Gets a collection of all grid job siblings. Job siblings are grid jobs that are executing within the same task.

      If task uses continuous mapper (i.e. it injected into task class) then job siblings will be requested from task node for each apply.

      Returns:
        Collection of grid job siblings executing within this task.
      Throws:
        `IgniteException` - If job siblings can not be received from task node.

    - #### refreshJobSiblings

          Collection<ComputeJobSibling> refreshJobSiblings()
                                                    throws IgniteException

      Refreshes collection of job siblings. This method has no effect when invoked on originating node, as the list of siblings is always most recent. However, when using `continuous mapping` (see `ComputeTaskContinuousMapper`), list of siblings on remote node may not be fresh. In that case, this method will re-request list of siblings from originating node.
      Returns:
        Refreshed collection of job siblings.
      Throws:
        `IgniteException` - If refresh failed.

    - #### getJobSibling

          @Nullable
          @Nullable ComputeJobSibling getJobSibling​(IgniteUuid jobId)
                                             throws IgniteException

      Gets job sibling for a given ID.

      If task uses continuous mapper (i.e. it injected into task class) then job sibling will be requested from task node for each apply.

      Parameters:
        `jobId` - Job ID to get the sibling for.
      Returns:
        Grid job sibling for a given ID.
      Throws:
        `IgniteException` - If job sibling can not be received from task node.

    - #### setAttribute

          void setAttribute​(Object key,
                            @Nullable
                            @Nullable Object val)
                     throws IgniteException

      Sets session attributed. Note that task session is distributed and this attribute will be propagated to all other jobs within this task and task itself - i.e., to all accessors of this session. Other jobs then will be notified by `ComputeTaskSessionAttributeListener` callback than an attribute has changed.

      This method is no-op if the session has finished.

      Parameters:
        `key` - Attribute key.
        `val` - Attribute value. Can be `null`.
      Throws:
        `IgniteException` - If sending of attribute message failed.

    - #### getAttribute

          @Nullable
          <K,​V> V getAttribute​(K key)

      Gets an attribute set by `setAttribute(Object, Object)` or `setAttributes(Map)` method. Note that this attribute could have been set by another job on another node.

      This method is no-op if the session has finished.

      Type Parameters:
        `K` - Attribute key type.
        `V` - Attribute value type.
      Parameters:
        `key` - Attribute key.
      Returns:
        Gets task attribute for given name.

    - #### setAttributes

          void setAttributes​(Map<?,​?> attrs)
                      throws IgniteException

      Sets task attributes. This method exists so one distributed replication operation will take place for the whole group of attributes passed in. Use it for performance reasons, rather than `setAttribute(Object, Object)` method, whenever you need to set multiple attributes.

      This method is no-op if the session has finished.

      Parameters:
        `attrs` - Attributes to set.
      Throws:
        `IgniteException` - If sending of attribute message failed.

    - #### getAttributes

          Map<?,​?> getAttributes()

      Gets all attributes.
      Returns:
        All session attributes.

    - #### addAttributeListener

          void addAttributeListener​(ComputeTaskSessionAttributeListener lsnr,
                                    boolean rewind)

      Add listener for the session attributes.
      Parameters:
        `lsnr` - Listener to add.
        `rewind` - `true` value will result in calling given listener for all already received attributes, while `false` value will result only in new attribute notification. Settings `rewind` to `true` allows for a simple mechanism that prevents the loss of notifications for the attributes that were previously received or received while this method was executing.

    - #### removeAttributeListener

          boolean removeAttributeListener​(ComputeTaskSessionAttributeListener lsnr)

      Removes given listener.
      Parameters:
        `lsnr` - Listener to remove.
      Returns:
        `true` if listener was removed, `false` otherwise.

    - #### waitForAttribute

          <K,​V> V waitForAttribute​(K key,
                                         long timeout)
                                  throws InterruptedException

      Waits for the specified attribute to be set. If this attribute is already in session this method will return immediately.
      Type Parameters:
        `K` - Attribute key type.
        `V` - Attribute value type.
      Parameters:
        `key` - Attribute key to wait for.
        `timeout` - Timeout in milliseconds to wait for. `0` means indefinite wait.
      Returns:
        Value of newly set attribute.
      Throws:
        `InterruptedException` - Thrown if wait was interrupted.

    - #### waitForAttribute

          <K,​V> boolean waitForAttribute​(K key,
                                               @Nullable
                                               V val,
                                               long timeout)
                                        throws InterruptedException

      Waits for the specified attribute to be set or updated with given value. Note that this method will block even if attribute is set for as long as its value is not equal to the specified.
      Type Parameters:
        `K` - Attribute key type.
        `V` - Attribute value type.
      Parameters:
        `key` - Attribute key to wait for.
        `val` - Attribute value to wait for. Can be `null`.
        `timeout` - Timeout in milliseconds to wait for. `0` means indefinite wait.
      Returns:
        Whether or not specified key/value pair has been set.
      Throws:
        `InterruptedException` - Thrown if wait was interrupted.

    - #### waitForAttributes

          Map<?,​?> waitForAttributes​(Collection<?> keys,
                                           long timeout)
                                    throws InterruptedException

      Waits for the specified attributes to be set. If these attributes are already in session this method will return immediately.
      Parameters:
        `keys` - Attribute keys to wait for.
        `timeout` - Timeout in milliseconds to wait for. `0` means indefinite wait.
      Returns:
        Attribute values mapped by their keys.
      Throws:
        `InterruptedException` - Thrown if wait was interrupted.

    - #### waitForAttributes

          boolean waitForAttributes​(Map<?,​?> attrs,
                                    long timeout)
                             throws InterruptedException

      Waits for the specified attributes to be set or updated with given values. Note that this method will block even if attributes are set for as long as their values are not equal to the specified.
      Parameters:
        `attrs` - Key/value pairs to wait for.
        `timeout` - Timeout in milliseconds to wait for. `0` means indefinite wait.
      Returns:
        Whether or not key/value pair has been set.
      Throws:
        `InterruptedException` - Thrown if wait was interrupted.

    - #### saveCheckpoint

          void saveCheckpoint​(String key,
                              Object state)
                       throws IgniteException

      Saves intermediate state of a job or task to a storage. The storage implementation is defined by `CheckpointSpi` implementation used.

      Long running jobs may decide to store intermediate state to protect themselves from failures. This way whenever a job fails over to another node, it can load its previously saved state via `loadCheckpoint(String)` method and continue with execution.

      This method defaults checkpoint scope to `ComputeTaskSessionScope.SESSION_SCOPE` and implementation will automatically remove the checkpoint at the end of the session. It is analogous to calling `saveCheckpoint(String, Serializable, GridCheckpointScope.SESSION_SCOPE, 0`.

      Parameters:
        `key` - Key to be used to load this checkpoint in future.
        `state` - Intermediate job state to save.
      Throws:
        `IgniteException` - If failed to save intermediate job state.
      See Also:
        `loadCheckpoint(String)`, `removeCheckpoint(String)`, `CheckpointSpi`

    - #### saveCheckpoint

          void saveCheckpoint​(String key,
                              Object state,
                              ComputeTaskSessionScope scope,
                              long timeout)
                       throws IgniteException

      Saves intermediate state of a job to a storage. The storage implementation is defined by `CheckpointSpi` implementation used.

      Long running jobs may decide to store intermediate state to protect themselves from failures. This way whenever a job fails over to another node, it can load its previously saved state via `loadCheckpoint(String)` method and continue with execution.

      The life time of the checkpoint is determined by its timeout and scope. If `ComputeTaskSessionScope.GLOBAL_SCOPE` is used, the checkpoint will outlive its session, and can only be removed by calling `CheckpointSpi.removeCheckpoint(String)` from `Ignite` or another task or job.

      Parameters:
        `key` - Key to be used to load this checkpoint in future.
        `state` - Intermediate job state to save.
        `scope` - Checkpoint scope. If equal to `ComputeTaskSessionScope.SESSION_SCOPE`, then state will automatically be removed at the end of task execution. Otherwise, if scope is `ComputeTaskSessionScope.GLOBAL_SCOPE` then state will outlive its session and can be removed by calling `removeCheckpoint(String)` from another task or whenever timeout expires.
        `timeout` - Maximum time this state should be kept by the underlying storage. Value `0` means that timeout will never expire.
      Throws:
        `IgniteException` - If failed to save intermediate job state.
      See Also:
        `loadCheckpoint(String)`, `removeCheckpoint(String)`, `CheckpointSpi`

    - #### saveCheckpoint

          void saveCheckpoint​(String key,
                              Object state,
                              ComputeTaskSessionScope scope,
                              long timeout,
                              boolean overwrite)
                       throws IgniteException

      Saves intermediate state of a job or task to a storage. The storage implementation is defined by `CheckpointSpi` implementation used.

      Long running jobs may decide to store intermediate state to protect themselves from failures. This way whenever a job fails over to another node, it can load its previously saved state via `loadCheckpoint(String)` method and continue with execution.

      The life time of the checkpoint is determined by its timeout and scope. If `ComputeTaskSessionScope.GLOBAL_SCOPE` is used, the checkpoint will outlive its session, and can only be removed by calling `CheckpointSpi.removeCheckpoint(String)` from `Ignite` or another task or job.

      Parameters:
        `key` - Key to be used to load this checkpoint in future.
        `state` - Intermediate job state to save.
        `scope` - Checkpoint scope. If equal to `ComputeTaskSessionScope.SESSION_SCOPE`, then state will automatically be removed at the end of task execution. Otherwise, if scope is `ComputeTaskSessionScope.GLOBAL_SCOPE` then state will outlive its session and can be removed by calling `removeCheckpoint(String)` from another task or whenever timeout expires.
        `timeout` - Maximum time this state should be kept by the underlying storage. Value `0` means that timeout will never expire.
        `overwrite` - Whether or not overwrite checkpoint if it already exists.
      Throws:
        `IgniteException` - If failed to save intermediate job state.
      See Also:
        `loadCheckpoint(String)`, `removeCheckpoint(String)`, `CheckpointSpi`

    - #### loadCheckpoint

          @Nullable
          <T> T loadCheckpoint​(String key)
                        throws IgniteException

      Loads job's state previously saved via `saveCheckpoint(String, Object, ComputeTaskSessionScope, long)` method from an underlying storage for a given `key`. If state was not previously saved, then `null` will be returned. The storage implementation is defined by `CheckpointSpi` implementation used.

      Long running jobs may decide to store intermediate state to protect themselves from failures. This way whenever a job starts, it can load its previously saved state and continue with execution.

      Type Parameters:
        `T` - Type of the checkpoint state.
      Parameters:
        `key` - Key for intermediate job state to load.
      Returns:
        Previously saved state or `null` if no state was found for a given `key`.
      Throws:
        `IgniteException` - If failed to load job state.
      See Also:
        `removeCheckpoint(String)`, `CheckpointSpi`

    - #### removeCheckpoint

          boolean removeCheckpoint​(String key)
                            throws IgniteException

      Removes previously saved job's state for a given `key` from an underlying storage. The storage implementation is defined by `CheckpointSpi` implementation used.

      Long running jobs may decide to store intermediate state to protect themselves from failures. This way whenever a job starts, it can load its previously saved state and continue with execution.

      Parameters:
        `key` - Key for intermediate job state to load.
      Returns:
        `true` if job state was removed, `false` if state was not found.
      Throws:
        `IgniteException` - If failed to remove job state.
      See Also:
        `loadCheckpoint(String)`, `CheckpointSpi`

    - #### getTopology

          Collection<UUID> getTopology()

      Gets a collection of grid nodes IDs.
      Returns:
        Collection of grid nodes IDs for the task's split.

    - #### mapFuture

          IgniteFuture<?> mapFuture()

      Gets future that will be completed when task "`map`" step has completed (which means that `ComputeTask.map(List, Object)` method has finished).
      Returns:
        Future that will be completed when task "`map`" step has completed.