/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.closure;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.binary.BinaryObjectException;
import org.apache.ignite.binary.BinaryRawReader;
import org.apache.ignite.binary.BinaryRawWriter;
import org.apache.ignite.binary.BinaryReader;
import org.apache.ignite.binary.BinaryWriter;
import org.apache.ignite.binary.Binarylizable;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.compute.ComputeJob;
import org.apache.ignite.compute.ComputeJobMasterLeaveAware;
import org.apache.ignite.compute.ComputeJobResult;
import org.apache.ignite.compute.ComputeJobResultPolicy;
import org.apache.ignite.compute.ComputeLoadBalancer;
import org.apache.ignite.compute.ComputeTaskSession;
import org.apache.ignite.internal.ComputeTaskInternalFuture;
import org.apache.ignite.internal.GridClosureCallMode;
import org.apache.ignite.internal.GridInternalWrapper;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.pool.PoolProcessor;
import org.apache.ignite.internal.processors.resource.GridNoImplicitInjection;
import org.apache.ignite.internal.util.GridSpinReadWriteLock;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.lang.GridPeerDeployAware;
import org.apache.ignite.internal.util.lang.GridPlainCallable;
import org.apache.ignite.internal.util.lang.GridPlainRunnable;
import org.apache.ignite.internal.util.lang.GridTuple3;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.internal.util.worker.GridWorkerFuture;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteReducer;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.resources.LoadBalancerResource;
import org.apache.ignite.thread.IgniteThread;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.compute.ComputeJobResultPolicy.FAILOVER;
import static org.apache.ignite.compute.ComputeJobResultPolicy.REDUCE;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_NO_FAILOVER;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SKIP_AUTH;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SUBGRID;
import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_TIMEOUT;

/**
 *
 */
public class GridClosureProcessor extends GridProcessorAdapter {
    /** Pool processor. */
    private final PoolProcessor pools;

    /** Lock to control execution after stop. */
    private final GridSpinReadWriteLock busyLock = new GridSpinReadWriteLock();

    /** Stopping flag. */
    private boolean stopping;

    /**
     * @param ctx Kernal context.
     */
    public GridClosureProcessor(GridKernalContext ctx) {
        super(ctx);

        pools = ctx.pools();

        assert pools != null;
    }

    /** {@inheritDoc} */
    @Override public void start() throws IgniteCheckedException {
        if (log.isDebugEnabled())
            log.debug("Started closure processor.");
    }

    /** {@inheritDoc} */
    @SuppressWarnings("BusyWait")
    @Override public void onKernalStop(boolean cancel) {
        boolean interrupted = false;

        // Busy wait is intentional.
        while (true) {
            try {
                if (busyLock.tryWriteLock(200, TimeUnit.MILLISECONDS))
                    break;
                else
                    Thread.sleep(200);
            }
            catch (InterruptedException ignore) {
                // Preserve interrupt status & ignore.
                // Note that interrupted flag is cleared.
                interrupted = true;
            }
        }

        try {
            if (interrupted)
                Thread.currentThread().interrupt();

            stopping = true;
        }
        finally {
            busyLock.writeUnlock();
        }

        if (log.isDebugEnabled())
            log.debug("Stopped closure processor.");
    }

    /**
     * @param mode Distribution mode.
     * @param jobs Closures to execute.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @return Task execution future.
     */
    public ComputeTaskInternalFuture<?> runAsync(GridClosureCallMode mode, @Nullable Collection<? extends Runnable> jobs,
        @Nullable Collection<ClusterNode> nodes, @Nullable String execName) {
        return runAsync(mode, jobs, nodes, false, execName);
    }

    /**
     * @param mode Distribution mode.
     * @param jobs Closures to execute.
     * @param nodes Grid nodes.
     * @param sys If {@code true}, then system pool will be used.
     * @param execName Custom executor name.
     * @return Task execution future.
     */
    public ComputeTaskInternalFuture<?> runAsync(GridClosureCallMode mode,
        Collection<? extends Runnable> jobs,
        @Nullable Collection<ClusterNode> nodes,
        boolean sys,
        @Nullable String execName)
    {
        assert mode != null;
        assert !F.isEmpty(jobs) : jobs;

        busyLock.readLock();

        try {
            if (stopping) {
                return ComputeTaskInternalFuture.finishedFuture(ctx, T1.class,
                    new IgniteCheckedException("Closure processor cannot be used on stopped grid: " + ctx.igniteInstanceName()));
            }

            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T1.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T1(mode, jobs), null, sys, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param mode Distribution mode.
     * @param job Closure to execute.
     * @param nodes Grid nodes.
     * @return Task execution future.
     */
    public ComputeTaskInternalFuture<?> runAsync(GridClosureCallMode mode, Runnable job,
        @Nullable Collection<ClusterNode> nodes) {
        return runAsync(mode, job, nodes, null);
    }

    /**
     * @param mode Distribution mode.
     * @param job Closure to execute.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @return Task execution future.
     */
    public ComputeTaskInternalFuture<?> runAsync(GridClosureCallMode mode, Runnable job,
        @Nullable Collection<ClusterNode> nodes, @Nullable String execName) {
        return runAsync(mode, job, nodes, false, execName);
    }

    /**
     * @param mode Distribution mode.
     * @param job Closure to execute.
     * @param nodes Grid nodes.
     * @param sys If {@code true}, then system pool will be used.
     * @param execName Custom executor name.
     * @return Task execution future.
     */
    public ComputeTaskInternalFuture<?> runAsync(GridClosureCallMode mode,
        Runnable job,
        @Nullable Collection<ClusterNode> nodes,
        boolean sys,
        @Nullable String execName)
    {
        assert mode != null;
        assert job != null;

        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T2.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T2(mode, job), null, sys, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * Maps {@link Runnable} jobs to specified nodes based on distribution mode.
     *
     * @param mode Distribution mode.
     * @param jobs Closures to map.
     * @param nodes Grid nodes.
     * @param lb Load balancer.
     * @throws IgniteException Thrown in case of any errors.
     * @return Mapping.
     */
    private Map<ComputeJob, ClusterNode> absMap(GridClosureCallMode mode, Collection<? extends Runnable> jobs,
        Collection<ClusterNode> nodes, ComputeLoadBalancer lb) throws IgniteException {
        assert mode != null;
        assert jobs != null;
        assert nodes != null;
        assert lb != null;

        try {
            if (!F.isEmpty(jobs) && !F.isEmpty(nodes)) {
                JobMapper mapper = new JobMapper(jobs.size());

                switch (mode) {
                    case BROADCAST: {
                        for (ClusterNode n : nodes)
                            for (Runnable r : jobs)
                                mapper.map(job(r), n);

                        break;
                    }

                    case BALANCE: {
                        for (Runnable r : jobs) {
                            ComputeJob job = job(r);

                            ClusterNode n = lb.getBalancedNode(job, null);

                            mapper.map(job, n);
                        }

                        break;
                    }
                }

                return mapper.map();
            }
            else
                return Collections.emptyMap();
        }
        catch (IgniteCheckedException e) {
            throw U.convertException(e);
        }
    }

    /**
     * Maps {@link Callable} jobs to specified nodes based on distribution mode.
     *
     * @param mode Distribution mode.
     * @param jobs Closures to map.
     * @param nodes Grid nodes.
     * @param lb Load balancer.
     * @throws IgniteException Thrown in case of any errors.
     * @return Mapping.
     */
    private <R> Map<ComputeJob, ClusterNode> outMap(GridClosureCallMode mode,
        Collection<? extends Callable<R>> jobs, Collection<ClusterNode> nodes, ComputeLoadBalancer lb)
        throws IgniteException {
        assert mode != null;
        assert jobs != null;
        assert nodes != null;
        assert lb != null;

        try {
            if (!F.isEmpty(jobs) && !F.isEmpty(nodes)) {
                JobMapper mapper = new JobMapper(jobs.size());

                switch (mode) {
                    case BROADCAST: {
                        for (ClusterNode n : nodes)
                            for (Callable<R> c : jobs)
                                mapper.map(job(c), n);

                        break;
                    }

                    case BALANCE: {
                        for (Callable<R> c : jobs) {
                            ComputeJob job = job(c);

                            ClusterNode n = lb.getBalancedNode(job, null);

                            mapper.map(job, n);
                        }

                        break;
                    }
                }

                return mapper.map();
            }
            else
                return Collections.emptyMap();
        }
        catch (IgniteCheckedException e) {
            throw U.convertException(e);
        }
    }

    /**
     * @param mode Distribution mode.
     * @param jobs Closures to execute.
     * @param rdc Reducer.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @param <R1> Type.
     * @param <R2> Type.
     * @return Reduced result.
     */
    public <R1, R2> ComputeTaskInternalFuture<R2> forkjoinAsync(GridClosureCallMode mode,
        Collection<? extends Callable<R1>> jobs,
        IgniteReducer<R1, R2> rdc,
        @Nullable Collection<ClusterNode> nodes,
        @Nullable String execName)
    {
        assert mode != null;
        assert rdc != null;
        assert !F.isEmpty(jobs);

        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T3.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T3<>(mode, jobs, rdc), null, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param mode Distribution mode.
     * @param jobs Closures to execute.
     * @param nodes Grid nodes.
     * @param <R> Type.
     * @return Grid future for collection of closure results.
     */
    public <R> ComputeTaskInternalFuture<Collection<R>> callAsync(
        GridClosureCallMode mode,
        @Nullable Collection<? extends Callable<R>> jobs,
        @Nullable Collection<ClusterNode> nodes) {
        return callAsync(mode, jobs, nodes, null);
    }

    /**
     * @param mode Distribution mode.
     * @param jobs Closures to execute.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @param <R> Type.
     * @return Grid future for collection of closure results.
     */
    public <R> ComputeTaskInternalFuture<Collection<R>> callAsync(
        GridClosureCallMode mode,
        @Nullable Collection<? extends Callable<R>> jobs,
        @Nullable Collection<ClusterNode> nodes,
        @Nullable String execName) {
        return callAsync(mode, jobs, nodes, false, execName);
    }

    /**
     * @param mode Distribution mode.
     * @param jobs Closures to execute.
     * @param nodes Grid nodes.
     * @param sys If {@code true}, then system pool will be used.
     * @param execName Custom executor name.
     * @param <R> Type.
     * @return Grid future for collection of closure results.
     */
    public <R> ComputeTaskInternalFuture<Collection<R>> callAsync(GridClosureCallMode mode,
        Collection<? extends Callable<R>> jobs,
        @Nullable Collection<ClusterNode> nodes,
        boolean sys,
        @Nullable String execName)
    {
        assert mode != null;
        assert !F.isEmpty(jobs);

        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T6.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T6<>(mode, jobs), null, sys, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param mode Distribution mode.
     * @param job Closure to execute.
     * @param nodes Grid nodes.
     * @param <R> Type.
     * @return Grid future for collection of closure results.
     */
    public <R> ComputeTaskInternalFuture<R> callAsync(GridClosureCallMode mode,
        @Nullable Callable<R> job, @Nullable Collection<ClusterNode> nodes) {
        return callAsync(mode, job, nodes, null);
    }

    /**
     * @param mode Distribution mode.
     * @param job Closure to execute.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @param <R> Type.
     * @return Grid future for collection of closure results.
     */
    public <R> ComputeTaskInternalFuture<R> callAsync(GridClosureCallMode mode,
        @Nullable Callable<R> job, @Nullable Collection<ClusterNode> nodes,
        @Nullable String execName) {
        return callAsync(mode, job, nodes, false, execName);
    }

    /**
     * @param cacheNames Cache names.
     * @param partId Partition.
     * @param job Closure to execute.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @return Grid future for collection of closure results.
     * @throws IgniteCheckedException If failed.
     */
    public <R> ComputeTaskInternalFuture<R> affinityCall(@NotNull Collection<String> cacheNames,
        int partId,
        Callable<R> job,
        @Nullable Collection<ClusterNode> nodes,
        @Nullable String execName) throws IgniteCheckedException {
        assert partId >= 0 : partId;

        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T5.class, U.emptyTopologyException());

            final String cacheName = F.first(cacheNames);

            final AffinityTopologyVersion mapTopVer = ctx.cache().context().exchange().readyAffinityVersion();
            final ClusterNode node = ctx.affinity().mapPartitionToNode(cacheName, partId, mapTopVer);

            if (node == null)
                return ComputeTaskInternalFuture.finishedFuture(ctx, T5.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T5(node, job, cacheNames, partId, mapTopVer), null,
                false, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param cacheNames Cache names.
     * @param partId Partition.
     * @param job Job.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @return Job future.
     * @throws IgniteCheckedException If failed.
     */
    public ComputeTaskInternalFuture<?> affinityRun(@NotNull Collection<String> cacheNames,
        int partId,
        Runnable job,
        @Nullable Collection<ClusterNode> nodes,
        @Nullable String execName) throws IgniteCheckedException {
        assert partId >= 0 : partId;

        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T4.class, U.emptyTopologyException());

            final String cacheName = F.first(cacheNames);

            final AffinityTopologyVersion mapTopVer = ctx.cache().context().exchange().readyAffinityVersion();
            final ClusterNode node = ctx.affinity().mapPartitionToNode(cacheName, partId, mapTopVer);

            if (node == null)
                return ComputeTaskInternalFuture.finishedFuture(ctx, T4.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T4(node, job, cacheNames, partId, mapTopVer), null,
                false, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param <R> Type.
     * @param mode Distribution mode.
     * @param job Closure to execute.
     * @param nodes Grid nodes.
     * @param sys If {@code true}, then system pool will be used.
     * @param timeout Timeout.
     * @param skipAuth Skip authorization check.
     * @return Grid future for collection of closure results.
     */
    public <R> IgniteInternalFuture<R> callAsyncNoFailover(
        GridClosureCallMode mode,
        @Nullable Callable<R> job,
        @Nullable Collection<ClusterNode> nodes,
        boolean sys,
        long timeout,
        boolean skipAuth) {
        assert mode != null;
        assert timeout >= 0 : timeout;

        busyLock.readLock();

        try {
            if (job == null)
                return new GridFinishedFuture<>();

            if (F.isEmpty(nodes))
                return new GridFinishedFuture<>(U.emptyTopologyException());

            ctx.task().setThreadContext(TC_NO_FAILOVER, true);
            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            if (skipAuth)
                ctx.task().setThreadContext(TC_SKIP_AUTH, true);

            if (timeout > 0)
                ctx.task().setThreadContext(TC_TIMEOUT, timeout);

            return ctx.task().execute(new T7<>(mode, job), null, sys);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param mode Distribution mode.
     * @param jobs Closures to execute.
     * @param nodes Grid nodes.
     * @param sys If {@code true}, then system pool will be used.
     * @param timeout If greater than 0 limits task execution. Cannot be negative.
     * @param <R> Type.
     * @return Grid future for collection of closure results.
     */
    public <R> IgniteInternalFuture<Collection<R>> callAsyncNoFailover(
        GridClosureCallMode mode,
        @Nullable Collection<? extends Callable<R>> jobs,
        @Nullable Collection<ClusterNode> nodes,
        boolean sys,
        long timeout
    ) {
        assert mode != null;
        assert timeout >= 0 : timeout;

        busyLock.readLock();

        try {
            if (F.isEmpty(jobs))
                return new GridFinishedFuture<>();

            if (F.isEmpty(nodes))
                return new GridFinishedFuture<>(U.emptyTopologyException());

            ctx.task().setThreadContext(TC_NO_FAILOVER, true);
            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            if (timeout > 0)
                ctx.task().setThreadContext(TC_TIMEOUT, timeout);

            return ctx.task().execute(new T6<>(mode, jobs), null, sys);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param mode Distribution mode.
     * @param job Closure to execute.
     * @param nodes Grid nodes.
     * @param sys If {@code true}, then system pool will be used.
     * @param execName Custom executor name.
     * @param <R> Type.
     * @return Grid future for collection of closure results.
     */
    public <R> ComputeTaskInternalFuture<R> callAsync(GridClosureCallMode mode,
        Callable<R> job,
        @Nullable Collection<ClusterNode> nodes,
        boolean sys,
        @Nullable String execName)
    {
        assert mode != null;
        assert job != null;

        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T7.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T7<>(mode, job), null, sys, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param job Job closure.
     * @param arg Optional job argument.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @return Grid future for execution result.
     */
    public <T, R> ComputeTaskInternalFuture<R> callAsync(IgniteClosure<T, R> job, @Nullable T arg,
        @Nullable Collection<ClusterNode> nodes, @Nullable String execName) {
        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T8.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T8(job, arg), null, false, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param job Job closure.
     * @param arg Optional job argument.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @return Grid future for execution result.
     */
    public <T, R> IgniteInternalFuture<Collection<R>> broadcast(IgniteClosure<T, R> job, @Nullable T arg,
        @Nullable Collection<ClusterNode> nodes, @Nullable String execName) {
        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return new GridFinishedFuture<>(U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T11<>(job), arg, false, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param job Job closure.
     * @param args Job arguments.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @return Grid future for execution result.
     */
    public <T, R> ComputeTaskInternalFuture<Collection<R>> callAsync(IgniteClosure<T, R> job,
        @Nullable Collection<? extends T> args,
        @Nullable Collection<ClusterNode> nodes,
        @Nullable String execName)
    {
        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T9.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T9<>(job, args), null, false, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param job Job closure.
     * @param args Job arguments.
     * @param rdc Reducer.
     * @param nodes Grid nodes.
     * @param execName Custom executor name.
     * @return Grid future for execution result.
     */
    public <T, R1, R2> ComputeTaskInternalFuture<R2> callAsync(IgniteClosure<T, R1> job,
        Collection<? extends T> args, IgniteReducer<R1, R2> rdc, @Nullable Collection<ClusterNode> nodes,
        @Nullable String execName) {
        busyLock.readLock();

        try {
            if (F.isEmpty(nodes))
                return ComputeTaskInternalFuture.finishedFuture(ctx, T10.class, U.emptyTopologyException());

            ctx.task().setThreadContext(TC_SUBGRID, nodes);

            return ctx.task().execute(new T10<>(job, args, rdc), null, false, execName);
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * @param thread Thread.
     * @param c Closure.
     */
    public void runLocalWithThreadPolicy(IgniteThread thread, Runnable c) {
        assert thread.stripe() >= 0 || thread.policy() != GridIoPolicy.UNDEFINED : thread;

        if (thread.stripe() >= 0)
            ctx.getStripedExecutorService().execute(thread.stripe(), c);
        else {
            try {
                ctx.pools().poolForPolicy(thread.policy()).execute(c);
            }
            catch (IgniteCheckedException e) {
                U.error(log, "Failed to get pool for policy: " + thread.policy(), e);
            }
        }
    }

    /**
     * @param c Closure to execute.
     * @param plc Whether to run on system or public pool.
     * @return Future.
     * @throws IgniteCheckedException Thrown in case of any errors.
     */
    public IgniteInternalFuture<?> runLocal(@Nullable final Runnable c, byte plc)
        throws IgniteCheckedException {
        if (c == null)
            return new GridFinishedFuture();

        busyLock.readLock();

        try {
            // Inject only if needed.
            if (!(c instanceof GridPlainRunnable))
                ctx.resource().inject(ctx.deploy().getDeployment(c.getClass().getName()), c.getClass(), c);

            final ClassLoader ldr = Thread.currentThread().getContextClassLoader();

            final GridWorkerFuture fut = new GridWorkerFuture();

            GridWorker w = new GridWorker(ctx.igniteInstanceName(), "closure-proc-worker", log) {
                @Override protected void body() {
                    try {
                        if (ldr != null)
                            U.wrapThreadLoader(ldr, c);
                        else
                            c.run();

                        fut.onDone();
                    }
                    catch (Throwable e) {
                        if (e instanceof Error)
                            U.error(log, "Closure execution failed with error.", e);

                        fut.onDone(U.cast(e));

                        if (e instanceof Error)
                            throw e;
                    }
                }
            };

            fut.setWorker(w);

            try {
                pools.poolForPolicy(plc).execute(w);
            }
            catch (RejectedExecutionException e) {
                U.error(log, "Failed to execute worker due to execution rejection " +
                    "(increase upper bound on executor service) [policy=" + plc + ']', e);

                w.run();
            }

            return fut;
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * Executes closure on system pool. In case of rejected execution re-runs the closure in the current
     * thread (blocking).
     *
     * @param c Closure to execute.
     * @return Future.
     */
    public IgniteInternalFuture<?> runLocalSafe(Runnable c) {
        return runLocalSafe(c, true);
    }

    /**
     * In case of rejected execution re-runs the closure in the current thread (blocking).
     *
     * @param c Closure to execute.
     * @param sys If {@code true}, then system pool will be used, otherwise public pool will be used.
     * @return Future.
     */
    public IgniteInternalFuture<?> runLocalSafe(Runnable c, boolean sys) {
        return runLocalSafe(c, sys ? GridIoPolicy.SYSTEM_POOL : GridIoPolicy.PUBLIC_POOL);
    }

    /**
     * In case of rejected execution re-runs the closure in the current thread (blocking).
     *
     * @param c Closure to execute.
     * @param plc Policy to choose executor pool.
     * @return Future.
     */
    public IgniteInternalFuture<?> runLocalSafe(Runnable c, byte plc) {
        try {
            return runLocal(c, plc);
        }
        catch (Throwable e) {
            if (e instanceof Error) {
                U.error(log, "Closure execution failed with error.", e);

                throw (Error)e;
            }

            // If execution was rejected - rerun locally.
            if (e.getCause() instanceof RejectedExecutionException) {
                U.warn(log, "Closure execution has been rejected (will execute in the same thread) [plc=" + plc +
                    ", closure=" + c + ']');

                try {
                    c.run();

                    return new GridFinishedFuture();
                }
                catch (Throwable t) {
                    if (t instanceof Error) {
                        U.error(log, "Closure execution failed with error.", t);

                        throw t;
                    }

                    return new GridFinishedFuture(U.cast(t));
                }
            }
            // If failed for other reasons - return error future.
            else
                return new GridFinishedFuture(U.cast(e));
        }
    }

    /**
     * @param c Closure to execute.
     * @param sys If {@code true}, then system pool will be used, otherwise public pool will be used.
     * @return Future.
     * @throws IgniteCheckedException Thrown in case of any errors.
     */
    private <R> IgniteInternalFuture<R> callLocal(@Nullable final Callable<R> c, boolean sys) throws IgniteCheckedException {
        return callLocal(c, sys ? GridIoPolicy.SYSTEM_POOL : GridIoPolicy.PUBLIC_POOL);
    }

    /**
     * @param c Closure to execute.
     * @param plc Whether to run on system or public pool.
     * @param <R> Type of closure return value.
     * @return Future.
     * @throws IgniteCheckedException Thrown in case of any errors.
     */
    public <R> IgniteInternalFuture<R> callLocal(@Nullable final Callable<R> c, byte plc)
        throws IgniteCheckedException {
        if (c == null)
            return new GridFinishedFuture<>();

        busyLock.readLock();

        try {
            // Inject only if needed.
            if (!(c instanceof GridPlainCallable))
                ctx.resource().inject(ctx.deploy().getDeployment(c.getClass().getName()), c.getClass(), c);

            final ClassLoader ldr = Thread.currentThread().getContextClassLoader();

            final GridWorkerFuture<R> fut = new GridWorkerFuture<>();

            GridWorker w = new GridWorker(ctx.igniteInstanceName(), "closure-proc-worker", log) {
                @Override protected void body() {
                    try {
                        if (ldr != null)
                            fut.onDone(U.wrapThreadLoader(ldr, c));
                        else
                            fut.onDone(c.call());
                    }
                    catch (Throwable e) {
                        if (e instanceof Error)
                            U.error(log, "Closure execution failed with error.", e);

                        fut.onDone(U.cast(e));

                        if (e instanceof Error)
                            throw (Error)e;
                    }
                }
            };

            fut.setWorker(w);

            try {
                pools.poolForPolicy(plc).execute(w);
            }
            catch (RejectedExecutionException e) {
                U.error(log, "Failed to execute worker due to execution rejection " +
                    "(increase upper bound on executor service) [policy=" + plc + ']', e);

                w.run();
            }

            return fut;
        }
        finally {
            busyLock.readUnlock();
        }
    }

    /**
     * Executes closure on system pool. Companion to {@link #callLocal(Callable, boolean)}
     * but in case of rejected execution re-runs the closure in the current thread (blocking).
     *
     * @param c Closure to execute.
     * @return Future.
     */
    public <R> IgniteInternalFuture<R> callLocalSafe(Callable<R> c) {
        return callLocalSafe(c, true);
    }

    /**
     * Executes closure on system pool. Companion to {@link #callLocal(Callable, boolean)}
     * but in case of rejected execution re-runs the closure in the current thread (blocking).
     *
     * @param c Closure to execute.
     * @param sys If {@code true}, then system pool will be used, otherwise public pool will be used.
     * @return Future.
     */
    public <R> IgniteInternalFuture<R> callLocalSafe(Callable<R> c, boolean sys) {
        return callLocalSafe(c, sys ? GridIoPolicy.SYSTEM_POOL : GridIoPolicy.PUBLIC_POOL);
    }

    /**
     * Companion to {@link #callLocal(Callable, boolean)} but in case of rejected execution re-runs
     * the closure in the current thread (blocking).
     *
     * @param c Closure to execute.
     * @param plc Policy to choose executor pool.
     * @return Future.
     */
    public  <R> IgniteInternalFuture<R> callLocalSafe(Callable<R> c, byte plc) {
        try {
            return callLocal(c, plc);
        }
        catch (IgniteCheckedException e) {
            // If execution was rejected - rerun locally.
            if (e.getCause() instanceof RejectedExecutionException) {
                U.warn(log, "Closure execution has been rejected (will execute in the same thread) [plc=" + plc +
                    ", closure=" + c + ']');

                try {
                    return new GridFinishedFuture<>(c.call());
                }
                // If failed again locally - return error future.
                catch (Exception e2) {
                    return new GridFinishedFuture<>(U.cast(e2));
                }
            }
            // If failed for other reasons - return error future.
            else
                return new GridFinishedFuture<>(U.cast(e));
        }
    }

    /**
     * Converts given closure with arguments to grid job.
     * @param job Job.
     * @param arg Optional argument.
     * @return Job.
     */
    private static <T, R> ComputeJob job(final IgniteClosure<T, R> job, @Nullable final T arg) {
        A.notNull(job, "job");

        return job instanceof ComputeJobMasterLeaveAware ? new C1MLA<>(job, arg) : new C1<>(job, arg);
    }

    /**
     * Converts given closure to a grid job.
     *
     * @param c Closure to convert to grid job.
     * @return Grid job made out of closure.
     */
    private static <R> ComputeJob job(final Callable<R> c) {
        A.notNull(c, "job");

        return c instanceof ComputeJobMasterLeaveAware ? new C2MLA<>(c) : new C2<>(c);
    }

    /**
     * Converts given closure to a grid job.
     *
     * @param r Closure to convert to grid job.
     * @return Grid job made out of closure.
     */
    private static ComputeJob job(final Runnable r) {
        A.notNull(r, "job");

        return r instanceof ComputeJobMasterLeaveAware ? new C4MLA(r) : new C4(r);
    }

    /**
     * Get collection of actual results.
     *
     * @param res Initial results.
     * @return Converted results.
     */
    private static <R> Collection<R> jobResults(List<ComputeJobResult> res) {
        if (res == null)
            return Collections.emptyList();
        else {
            List<R> res0 = new ArrayList<>(res.size());

            for (int i = 0; i < res.size(); i++)
                res0.add(res.get(i).<R>getData());

            return res0;
        }
    }

    /**
     *
     */
    private class JobMapper {
        /** */
        private final Map<ComputeJob, ClusterNode> map;

        /** */
        private boolean hadLocNode;

        /**
         * @param expJobCnt Expected Jobs count.
         */
        private JobMapper(int expJobCnt) {
            map = IgniteUtils.newHashMap(expJobCnt);
        }

        /**
         * @param job Job.
         * @param node Node.
         * @throws IgniteCheckedException In case of error.
         */
        public void map(@NotNull ComputeJob job, @NotNull ClusterNode node) throws IgniteCheckedException {
            if (ctx.localNodeId().equals(node.id())) {
                if (hadLocNode) {
                    Marshaller marsh = ctx.config().getMarshaller();

                    job = U.unmarshal(marsh, U.marshal(marsh, job), U.resolveClassLoader(ctx.config()));
                }
                else
                    hadLocNode = true;
            }

            map.put(job, node);
        }

        /**
         * @return Map.
         */
        public Map<ComputeJob, ClusterNode> map() {
            return map;
        }
    }

    /**
     * No-reduce task adapter.
     */
    private abstract static class TaskNoReduceAdapter<T> extends GridPeerDeployAwareTaskAdapter<T, Void> {
        /** */
        private static final long serialVersionUID = 0L;

        /**
         * @param pda Peer deploy aware instance.
         */
        protected TaskNoReduceAdapter(@Nullable GridPeerDeployAware pda) {
            super(pda);
        }

        /** {@inheritDoc} */
        @Nullable @Override public Void reduce(List<ComputeJobResult> results) {
            return null;
        }
    }

    /**
     * Task that is free of dragged in enclosing context for the method
     * {@link GridClosureProcessor#runAsync(GridClosureCallMode, Collection, Collection,String)}.
     */
    private class T1 extends TaskNoReduceAdapter<Void> implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        @LoadBalancerResource
        private ComputeLoadBalancer lb;

        /** */
        private IgniteBiTuple<GridClosureCallMode, Collection<? extends Runnable>> t;

        /**
         * @param mode Call mode.
         * @param jobs Collection of jobs.
         */
        private T1(GridClosureCallMode mode, Collection<? extends Runnable> jobs) {
            super(U.peerDeployAware0(jobs));

            t = F.<
                GridClosureCallMode,
                Collection<? extends Runnable>
                >t(mode, jobs);
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            return absMap(t.get1(), t.get2(), subgrid, lb);
        }
    }

    /**
     * Task that is free of dragged in enclosing context for the method
     * {@link GridClosureProcessor#runAsync(GridClosureCallMode, Runnable, Collection, String)}.
     */
    private class T2 extends TaskNoReduceAdapter<Void> implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        @LoadBalancerResource
        private ComputeLoadBalancer lb;

        /** */
        private IgniteBiTuple<GridClosureCallMode, Runnable> t;

        /**
         * @param mode Call mode.
         * @param job Job.
         */
        private T2(GridClosureCallMode mode, Runnable job) {
            super(U.peerDeployAware(job));

            t = F.t(mode, job);
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            return absMap(t.get1(), F.asList(t.get2()), subgrid, lb);
        }
    }

    /**
     * Task that is free of dragged in enclosing context for the method
     * {@link GridClosureProcessor#forkjoinAsync(GridClosureCallMode, Collection, org.apache.ignite.lang.IgniteReducer, Collection, String)}
     */
    private class T3<R1, R2> extends GridPeerDeployAwareTaskAdapter<Void, R2> implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        @LoadBalancerResource
        private ComputeLoadBalancer lb;

        /** */
        private GridTuple3<GridClosureCallMode,
            Collection<? extends Callable<R1>>,
            IgniteReducer<R1, R2>
            > t;

        /**
         *
         * @param mode Call mode.
         * @param jobs Collection of jobs.
         * @param rdc Reducer.
         */
        private T3(GridClosureCallMode mode, Collection<? extends Callable<R1>> jobs, IgniteReducer<R1, R2> rdc) {
            super(U.peerDeployAware0(jobs));

            t = F.<
                GridClosureCallMode,
                Collection<? extends Callable<R1>>,
                IgniteReducer<R1, R2>
                >t(mode, jobs, rdc);
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            return outMap(t.get1(), t.get2(), subgrid, lb);
        }

        /** {@inheritDoc} */
        @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) {
            ComputeJobResultPolicy resPlc = super.result(res, rcvd);

            if (res.getException() == null && resPlc != FAILOVER && !t.get3().collect((R1)res.getData()))
                resPlc = REDUCE; // If reducer returned false - reduce right away.

            return resPlc;
        }

        /** {@inheritDoc} */
        @Override public R2 reduce(List<ComputeJobResult> res) {
            return t.get3().reduce();
        }
    }

    /**
     */
    private static class T4 extends TaskNoReduceAdapter<Void> implements GridNoImplicitInjection, AffinityTask {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private ClusterNode node;

        /** */
        private Runnable job;

        /** */
        private int partId;

        /** */
        private AffinityTopologyVersion topVer;

        /** */
        private Collection<String> affCacheNames;

        /**
         * @param node Cluster node.
         * @param job Job affinity partition.
         * @param affCacheNames Affinity caches.
         * @param partId Partition.
         * @param topVer Affinity topology version.
         */
        private T4(ClusterNode node, Runnable job, Collection<String> affCacheNames, int partId,
            AffinityTopologyVersion topVer) {
            super(U.peerDeployAware0(job));

            assert partId >= 0;

            this.node = node;
            this.job = job;
            this.affCacheNames = affCacheNames;
            this.partId = partId;
            this.topVer = topVer;
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            return Collections.singletonMap(job(job), node);
        }

        /** {@inheritDoc} */
        @Override public int partition() {
            return partId;
        }

        /** {@inheritDoc} */
        @Nullable @Override public Collection<String> affinityCacheNames() {
            return affCacheNames;
        }

        /** {@inheritDoc} */
        @Nullable @Override public AffinityTopologyVersion topologyVersion() {
            return topVer;
        }
    }

    /**
     */
    private static class T5<R> extends GridPeerDeployAwareTaskAdapter<Void, R> implements
        GridNoImplicitInjection, AffinityTask {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private ClusterNode node;

        /** */
        private Callable<R> job;

        /** */
        private int partId;

        /** */
        private AffinityTopologyVersion topVer;

        /** */
        private Collection<String> affCacheNames;

        /**
         * @param node Cluster node.
         * @param job Job affinity partition.
         * @param affCacheNames Affinity caches.
         * @param partId Partition.
         * @param topVer Affinity topology version.
         */
        private T5(ClusterNode node,
            Callable<R> job,
            Collection<String> affCacheNames,
            int partId,
            AffinityTopologyVersion topVer) {
            super(U.peerDeployAware0(job));

            this.node = node;
            this.job = job;
            this.affCacheNames = affCacheNames;
            this.partId = partId;
            this.topVer = topVer;
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            return Collections.singletonMap(job(job), node);
        }

        /** {@inheritDoc} */
        @Override public R reduce(List<ComputeJobResult> res) {
            for (ComputeJobResult r : res) {
                if (r.getException() == null)
                    return r.getData();
            }

            throw new IgniteException("Failed to find successful job result: " + res);
        }

        /** {@inheritDoc} */
        @Override public int partition() {
            return partId;
        }

        /** {@inheritDoc} */
        @Nullable @Override public Collection<String> affinityCacheNames() {
            return affCacheNames;
        }

        /** {@inheritDoc} */
        @Nullable @Override public AffinityTopologyVersion topologyVersion() {
            return topVer;
        }
    }

    /**
     * Task that is free of dragged in enclosing context for the method
     * {@link GridClosureProcessor#callAsync(GridClosureCallMode, Collection, Collection, String)}
     */
    private class T6<R> extends GridPeerDeployAwareTaskAdapter<Void, Collection<R>> implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private final GridClosureCallMode mode;

        /** */
        private final Collection<? extends Callable<R>> jobs;

        /** */
        @LoadBalancerResource
        private ComputeLoadBalancer lb;

        /**
         *
         * @param mode Call mode.
         * @param jobs Collection of jobs.
         */
        private T6(
            GridClosureCallMode mode,
            Collection<? extends Callable<R>> jobs) {
            super(U.peerDeployAware0(jobs));

            this.mode = mode;
            this.jobs = jobs;
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            return outMap(mode, jobs, subgrid, lb);
        }

        /** {@inheritDoc} */
        @Override public Collection<R> reduce(List<ComputeJobResult> res) {
            return jobResults(res);
        }
    }

    /**
     * Task that is free of dragged in enclosing context for the method
     * {@link GridClosureProcessor#callAsync(GridClosureCallMode, Callable, Collection, String)}
     */
    private class T7<R> extends GridPeerDeployAwareTaskAdapter<Void, R> implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private IgniteBiTuple<GridClosureCallMode, Callable<R>> t;

        /** */
        @LoadBalancerResource
        private ComputeLoadBalancer lb;

        /**
         * @param mode Call mode.
         * @param job Job.
         */
        private T7(GridClosureCallMode mode, Callable<R> job) {
            super(U.peerDeployAware(job));

            t = F.t(mode, job);
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            return outMap(t.get1(), F.asList(t.get2()), subgrid, lb);
        }

        /** {@inheritDoc} */
        @Override public R reduce(List<ComputeJobResult> res) {
            for (ComputeJobResult r : res)
                if (r.getException() == null)
                    return r.getData();

            throw new IgniteException("Failed to find successful job result: " + res);
        }
    }

    /**
     */
    private static class T8<T, R> extends GridPeerDeployAwareTaskAdapter<Void, R> implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private IgniteClosure<T, R> job;

        /** */
        private T arg;

        /** */
        @LoadBalancerResource
        private ComputeLoadBalancer lb;

        /**
         * @param job Job.
         * @param arg Optional job argument.
         */
        private T8(IgniteClosure<T, R> job, @Nullable T arg) {
            super(U.peerDeployAware(job));

            this.job = job;
            this.arg = arg;
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            ComputeJob job = job(this.job, this.arg);

            ClusterNode node = lb.getBalancedNode(job, null);

            return Collections.singletonMap(job, node);
        }

        /** {@inheritDoc} */
        @Override public R reduce(List<ComputeJobResult> res) throws IgniteException {
            for (ComputeJobResult r : res)
                if (r.getException() == null)
                    return r.getData();

            throw new IgniteException("Failed to find successful job result: " + res);
        }
    }

    /**
     */
    private class T9<T, R> extends GridPeerDeployAwareTaskAdapter<Void, Collection<R>>
        implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private IgniteClosure<T, R> job;

        /** */
        private Collection<? extends T> args;

        /** */
        @LoadBalancerResource
        private ComputeLoadBalancer lb;

        /**
         * @param job Job.
         * @param args Job arguments.
         */
        private T9(IgniteClosure<T, R> job, Collection<? extends T> args) {
            super(U.peerDeployAware(job));

            this.job = job;
            this.args = args;
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            try {
                JobMapper mapper = new JobMapper(args.size());

                for (T jobArg : args) {
                    ComputeJob job = job(this.job, jobArg);

                    ClusterNode node = lb.getBalancedNode(job, null);

                    mapper.map(job, node);
                }

                return mapper.map();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public Collection<R> reduce(List<ComputeJobResult> res) throws IgniteException {
            return jobResults(res);
        }
    }

    /**
     */
    private class T10<T, R1, R2> extends GridPeerDeployAwareTaskAdapter<Void, R2> implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private IgniteClosure<T, R1> job;

        /** */
        private Collection<? extends T> args;

        /** */
        private IgniteReducer<R1, R2> rdc;

        /** */
        @LoadBalancerResource
        private ComputeLoadBalancer lb;

        /**
         * @param job Job.
         * @param args Job arguments.
         * @param rdc Reducer.
         */
        private T10(IgniteClosure<T, R1> job, Collection<? extends T> args, IgniteReducer<R1, R2> rdc) {
            super(U.peerDeployAware(job));

            this.job = job;
            this.args = args;
            this.rdc = rdc;
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Void arg) {
            try {
                JobMapper mapper = new JobMapper(args.size());

                for (T jobArg : args) {
                    ComputeJob job = job(this.job, jobArg);

                    ClusterNode node = lb.getBalancedNode(job, null);

                    mapper.map(job, node);
                }

                return mapper.map();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) {
            ComputeJobResultPolicy resPlc = super.result(res, rcvd);

            if (res.getException() == null && resPlc != FAILOVER && !rdc.collect((R1)res.getData()))
                resPlc = REDUCE; // If reducer returned false - reduce right away.

            return resPlc;
        }

        /** {@inheritDoc} */
        @Override public R2 reduce(List<ComputeJobResult> res) {
            return rdc.reduce();
        }
    }

    /**
     */
    private class T11<T, R> extends GridPeerDeployAwareTaskAdapter<T, Collection<R>>
        implements GridNoImplicitInjection {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private final IgniteClosure<T, R> job;

        /**
         * @param job Job.
         */
        private T11(IgniteClosure<T, R> job) {
            super(U.peerDeployAware(job));

            this.job = job;
        }

        /** {@inheritDoc} */
        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable T arg) {
            if (F.isEmpty(subgrid))
                return Collections.emptyMap();

            try {
                JobMapper mapper = new JobMapper(subgrid.size());

                for (ClusterNode n : subgrid)
                    mapper.map(job(job, arg), n);

                return mapper.map();
            }
            catch (IgniteCheckedException e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public Collection<R> reduce(List<ComputeJobResult> res) {
            return jobResults(res);
        }
    }

    /**
     *
     */
    public static class C1<T, R> implements ComputeJob, Binarylizable, GridNoImplicitInjection,
        GridInternalWrapper<IgniteClosure> {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        @GridToStringInclude
        protected IgniteClosure<T, R> job;

        /** */
        @GridToStringInclude(sensitive = true)
        protected T arg;

        /**
         *
         */
        public C1() {
            // No-op.
        }

        /**
         * @param job Job.
         * @param arg Argument.
         */
        C1(IgniteClosure<T, R> job, T arg) {
            this.job = job;
            this.arg = arg;
        }

        /** {@inheritDoc} */
        @Nullable @Override public Object execute() {
            return job.apply(arg);
        }

        /** {@inheritDoc} */
        @Override public void cancel() {
            // No-op.
        }

        @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException {
            BinaryRawWriter rawWriter = writer.rawWriter();

            rawWriter.writeObject(job);
            rawWriter.writeObject(arg);
        }

        @Override public void readBinary(BinaryReader reader) throws BinaryObjectException {
            BinaryRawReader rawReader = reader.rawReader();

            job = rawReader.readObject();
            arg = rawReader.readObject();
        }

        /** {@inheritDoc} */
        @Override public IgniteClosure userObject() {
            return job;
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(C1.class, this);
        }
    }

    /**
     *
     */
    public static class C1MLA<T, R> extends C1<T, R> implements ComputeJobMasterLeaveAware {
        /** */
        private static final long serialVersionUID = 0L;

        /**
         *
         */
        public C1MLA() {
            // No-op.
        }

        /**
         * @param job Job.
         * @param arg Argument.
         */
        private C1MLA(IgniteClosure<T, R> job, T arg) {
            super(job, arg);
        }

        /** {@inheritDoc} */
        @Override public void onMasterNodeLeft(ComputeTaskSession ses) {
            ((ComputeJobMasterLeaveAware)job).onMasterNodeLeft(ses);
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(C1MLA.class, this, super.toString());
        }
    }

    /**
     *
     */
    public static class C2<R> implements ComputeJob, Binarylizable, GridNoImplicitInjection,
        GridInternalWrapper<Callable> {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        @GridToStringInclude
        protected Callable<R> c;

        /**
         *
         */
        public C2() {
            // No-op.
        }

        /**
         * @param c Callable.
         */
        private C2(Callable<R> c) {
            this.c = c;
        }

        /** {@inheritDoc} */
        @Override public Object execute() {
            try {
                return c.call();
            }
            catch (Exception e) {
                throw new IgniteException(e);
            }
        }

        /** {@inheritDoc} */
        @Override public void cancel() {
            // No-op.
        }

        @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException {
            writer.rawWriter().writeObject(c);
        }

        @Override public void readBinary(BinaryReader reader) throws BinaryObjectException {
            c = reader.rawReader().readObject();
        }

        /** {@inheritDoc} */
        @Override public Callable userObject() {
            return c;
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(C2.class, this);
        }
    }

    /**
     *
     */
    public static class C2MLA<R> extends C2<R> implements ComputeJobMasterLeaveAware {
        /** */
        private static final long serialVersionUID = 0L;

        /**
         *
         */
        public C2MLA() {
            // No-op.
        }

        /**
         * @param c Callable.
         */
        private C2MLA(Callable<R> c) {
            super(c);
        }

        /** {@inheritDoc} */
        @Override public void onMasterNodeLeft(ComputeTaskSession ses) {
            ((ComputeJobMasterLeaveAware)c).onMasterNodeLeft(ses);
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(C2MLA.class, this, super.toString());
        }
    }

    /**
     */
    public static class C4 implements ComputeJob, Binarylizable, GridNoImplicitInjection, GridInternalWrapper<Runnable> {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        @GridToStringInclude
        protected Runnable r;

        /**
         *
         */
        public C4() {
            // No-op.
        }

        /**
         * @param r Runnable.
         */
        private C4(Runnable r) {
            this.r = r;
        }

        /** {@inheritDoc} */
        @Nullable @Override public Object execute() {
            r.run();

            return null;
        }

        /** {@inheritDoc} */
        @Override public void cancel() {
            // No-op.
        }

        @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException {
            writer.rawWriter().writeObject(r);
        }

        @Override public void readBinary(BinaryReader reader) throws BinaryObjectException {
            r = reader.rawReader().readObject();
        }

        /** {@inheritDoc} */
        @Override public Runnable userObject() {
            return r;
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(C4.class, this);
        }
    }

    /**
     *
     */
    public static class C4MLA extends C4 implements ComputeJobMasterLeaveAware {
        /** */
        private static final long serialVersionUID = 0L;

        /**
         *
         */
        public C4MLA() {
            // No-op.
        }

        /**
         * @param r Runnable.
         */
        private C4MLA(Runnable r) {
            super(r);
        }

        /** {@inheritDoc} */
        @Override public void onMasterNodeLeft(ComputeTaskSession ses) {
            ((ComputeJobMasterLeaveAware)r).onMasterNodeLeft(ses);
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(C4MLA.class, this, super.toString());
        }
    }
}