/*
 * 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.service;

import java.lang.Thread.UncaughtExceptionHandler;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import javax.cache.Cache;
import javax.cache.event.CacheEntryEvent;
import javax.cache.event.CacheEntryUpdatedListener;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.compute.ComputeJobContext;
import org.apache.ignite.configuration.DeploymentMode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.GridClosureCallMode;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.managers.eventstorage.DiscoveryEventListener;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
import org.apache.ignite.internal.processors.cache.CacheIteratorConverter;
import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessage;
import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateProposedMessage;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.query.CacheQuery;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import org.apache.ignite.internal.processors.cluster.IgniteChangeGlobalStateSupport;
import org.apache.ignite.internal.processors.task.GridInternal;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.util.GridEmptyIterator;
import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.GridCloseableIterator;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.security.SecurityException;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.resources.JobContextResource;
import org.apache.ignite.resources.LoggerResource;
import org.apache.ignite.services.Service;
import org.apache.ignite.services.ServiceConfiguration;
import org.apache.ignite.services.ServiceDeploymentException;
import org.apache.ignite.services.ServiceDescriptor;
import org.apache.ignite.thread.IgniteThreadFactory;
import org.apache.ignite.thread.OomExceptionHandler;
import org.apache.ignite.transactions.Transaction;
import org.jetbrains.annotations.Nullable;

import static javax.cache.event.EventType.REMOVED;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_SERVICES_COMPATIBILITY_MODE;
import static org.apache.ignite.IgniteSystemProperties.getString;
import static org.apache.ignite.configuration.DeploymentMode.ISOLATED;
import static org.apache.ignite.configuration.DeploymentMode.PRIVATE;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_SERVICES_COMPATIBILITY_MODE;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;

/**
 * Grid service processor.
 */
@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter", "ConstantConditions"})
public class GridServiceProcessor extends GridProcessorAdapter implements IgniteChangeGlobalStateSupport {
    /** */
    private final Boolean srvcCompatibilitySysProp;

    /** Time to wait before reassignment retries. */
    private static final long RETRY_TIMEOUT = 1000;

    /** */
    private static final int[] EVTS = {
        EventType.EVT_NODE_JOINED,
        EventType.EVT_NODE_LEFT,
        EventType.EVT_NODE_FAILED,
        DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT
    };

    /** Local service instances. */
    private final Map<String, Collection<ServiceContextImpl>> locSvcs = new HashMap<>();

    /** Deployment futures. */
    private final ConcurrentMap<String, GridServiceDeploymentFuture> depFuts = new ConcurrentHashMap<>();

    /** Deployment futures. */
    private final ConcurrentMap<String, GridFutureAdapter<?>> undepFuts = new ConcurrentHashMap<>();

    /** Pending compute job contexts that waiting for utility cache initialization. */
    private final List<ComputeJobContext> pendingJobCtxs = new ArrayList<>(0);

    /** Deployment executor service. */
    private volatile ExecutorService depExe;

    /** Busy lock. */
    private volatile GridSpinBusyLock busyLock = new GridSpinBusyLock();

    /** Uncaught exception handler for thread pools. */
    private final UncaughtExceptionHandler oomeHnd = new OomExceptionHandler(ctx);

    /** Thread factory. */
    private ThreadFactory threadFactory = new IgniteThreadFactory(ctx.igniteInstanceName(), "service",
        oomeHnd);

    /** Thread local for service name. */
    private ThreadLocal<String> svcName = new ThreadLocal<>();

    /** Service cache. */
    private volatile IgniteInternalCache<Object, Object> serviceCache;

    /** Topology listener. */
    private DiscoveryEventListener topLsnr = new TopologyListener();

    /** */
    private final CountDownLatch startLatch = new CountDownLatch(1);

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

        depExe = Executors.newSingleThreadExecutor(new IgniteThreadFactory(ctx.igniteInstanceName(),
            "srvc-deploy", oomeHnd));

        String servicesCompatibilityMode = getString(IGNITE_SERVICES_COMPATIBILITY_MODE);

        srvcCompatibilitySysProp = servicesCompatibilityMode == null ? null : Boolean.valueOf(servicesCompatibilityMode);
    }

    /**
     * @param ctx Context.
     * @throws IgniteCheckedException If failed.
     */
    public void onContinuousProcessorStarted(GridKernalContext ctx) throws IgniteCheckedException {
        if (ctx.clientNode()) {
            assert !ctx.isDaemon();

            ctx.continuous().registerStaticRoutine(
                CU.UTILITY_CACHE_NAME, new ServiceEntriesListener(), null, null
            );
        }
    }

    /** {@inheritDoc} */
    @Override public void start() throws IgniteCheckedException {
        ctx.addNodeAttribute(ATTR_SERVICES_COMPATIBILITY_MODE, srvcCompatibilitySysProp);

        if (ctx.isDaemon())
            return;

        IgniteConfiguration cfg = ctx.config();

        DeploymentMode depMode = cfg.getDeploymentMode();

        if (cfg.isPeerClassLoadingEnabled() && (depMode == PRIVATE || depMode == ISOLATED) &&
            !F.isEmpty(cfg.getServiceConfiguration()))
            throw new IgniteCheckedException("Cannot deploy services in PRIVATE or ISOLATED deployment mode: " + depMode);
    }

    /** {@inheritDoc} */
    @SuppressWarnings("unchecked")
    @Override public void onKernalStart(boolean active) throws IgniteCheckedException {
        if (ctx.isDaemon() || !active)
            return;

        onKernalStart0();
    }

    /**
     * Do kernal start.
     *
     * @throws IgniteCheckedException If failed.
     */
    private void onKernalStart0() throws IgniteCheckedException {
        if (!ctx.clientNode())
            ctx.event().addDiscoveryEventListener(topLsnr, EVTS);

        updateUtilityCache();

        startLatch.countDown();

        try {
            if (ctx.deploy().enabled())
                ctx.cache().context().deploy().ignoreOwnership(true);

            if (!ctx.clientNode()) {
                DiscoveryDataClusterState clusterState = ctx.state().clusterState();

                boolean isLocLsnr = !clusterState.hasBaselineTopology() ||
                    CU.baselineNode(ctx.cluster().get().localNode(), clusterState);

                // Register query listener and run it for local entries, if data is available locally.
                // It is also invoked on rebalancing.
                // Otherwise remote listener is registered.
                serviceCache.context().continuousQueries().executeInternalQuery(
                    new ServiceEntriesListener(), null, isLocLsnr, true, false, false
                );
            }
            else { // Listener for client nodes is registered in onContinuousProcessorStarted method.
                assert !ctx.isDaemon();

                ctx.closure().runLocalSafe(new Runnable() {
                    @Override public void run() {
                        try {
                            Iterable<CacheEntryEvent<?, ?>> entries =
                                serviceCache.context().continuousQueries().existingEntries(false, null);

                            onSystemCacheUpdated(entries);
                        }
                        catch (IgniteCheckedException e) {
                            U.error(log, "Failed to load service entries: " + e, e);
                        }
                    }
                });
            }
        }
        finally {
            if (ctx.deploy().enabled())
                ctx.cache().context().deploy().ignoreOwnership(false);
        }

        ServiceConfiguration[] cfgs = ctx.config().getServiceConfiguration();

        if (cfgs != null)
            deployAll(Arrays.asList(cfgs), ctx.cluster().get().forServers().predicate()).get();

        if (log.isDebugEnabled())
            log.debug("Started service processor.");
    }

    /**
     *
     */
    public void updateUtilityCache() {
        serviceCache = ctx.cache().utilityCache();
    }

    /**
     * @return Service cache.
     */
    private IgniteInternalCache<Object, Object> serviceCache() {
        if (serviceCache == null)
            U.awaitQuiet(startLatch);

        return serviceCache;
    }

    /** {@inheritDoc} */
    @Override public void onKernalStop(boolean cancel) {
        if (ctx.isDaemon())
            return;

        GridSpinBusyLock busyLock = this.busyLock;

        // Will not release it.
        if (busyLock != null) {
            busyLock.block();

            this.busyLock = null;
        }

        startLatch.countDown();

        U.shutdownNow(GridServiceProcessor.class, depExe, log);

        if (!ctx.clientNode())
            ctx.event().removeDiscoveryEventListener(topLsnr);

        Collection<ServiceContextImpl> ctxs = new ArrayList<>();

        synchronized (locSvcs) {
            for (Collection<ServiceContextImpl> ctxs0 : locSvcs.values())
                ctxs.addAll(ctxs0);

            locSvcs.clear();
        }

        for (ServiceContextImpl ctx : ctxs) {
            ctx.setCancelled(true);

            Service svc = ctx.service();

            if (svc != null)
                try {
                    svc.cancel(ctx);
                }
                catch (Throwable e) {
                    log.error("Failed to cancel service (ignoring) [name=" + ctx.name() +
                        ", execId=" + ctx.executionId() + ']', e);

                    if (e instanceof Error)
                        throw e;
                }

            ctx.executor().shutdownNow();
        }

        for (ServiceContextImpl ctx : ctxs) {
            try {
                if (log.isInfoEnabled() && !ctxs.isEmpty())
                    log.info("Shutting down distributed service [name=" + ctx.name() + ", execId8=" +
                        U.id8(ctx.executionId()) + ']');

                ctx.executor().awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
            }
            catch (InterruptedException ignore) {
                Thread.currentThread().interrupt();

                U.error(log, "Got interrupted while waiting for service to shutdown (will continue stopping node): " +
                    ctx.name());
            }
        }

        Exception err = new IgniteCheckedException("Operation has been cancelled (node is stopping).");

        cancelFutures(depFuts, err);
        cancelFutures(undepFuts, err);

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

    /** {@inheritDoc} */
    @Override public void onActivate(GridKernalContext kctx) throws IgniteCheckedException {
        if (log.isDebugEnabled())
            log.debug("Activate service processor [nodeId=" + ctx.localNodeId() +
                " topVer=" + ctx.discovery().topologyVersionEx() + " ]");

        busyLock = new GridSpinBusyLock();

        depExe = Executors.newSingleThreadExecutor(new IgniteThreadFactory(ctx.igniteInstanceName(),
            "srvc-deploy", oomeHnd));

        start();

        onKernalStart0();
    }

    /** {@inheritDoc} */
    @Override public void onDeActivate(GridKernalContext kctx) {
        if (log.isDebugEnabled())
            log.debug("DeActivate service processor [nodeId=" + ctx.localNodeId() +
                " topVer=" + ctx.discovery().topologyVersionEx() + " ]");

        cancelFutures(depFuts, new IgniteCheckedException("Failed to deploy service, cluster in active."));

        cancelFutures(undepFuts, new IgniteCheckedException("Failed to undeploy service, cluster in active."));

        onKernalStop(true);
    }

    /** {@inheritDoc} */
    @Override public void onDisconnected(IgniteFuture<?> reconnectFut) throws IgniteCheckedException {
        cancelFutures(depFuts, new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
            "Failed to deploy service, client node disconnected."));

        cancelFutures(undepFuts, new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
            "Failed to undeploy service, client node disconnected."));
    }

    /**
     * @param futs Futs.
     * @param err Exception.
     */
    private void cancelFutures(ConcurrentMap<String, ? extends GridFutureAdapter<?>> futs, Exception err) {
        for (Map.Entry<String, ? extends GridFutureAdapter<?>> entry : futs.entrySet()) {
            GridFutureAdapter fut = entry.getValue();

            fut.onDone(err);

            futs.remove(entry.getKey(), fut);
        }
    }

    /**
     * Validates service configuration.
     *
     * @param c Service configuration.
     * @throws IgniteException If validation failed.
     */
    private void validate(ServiceConfiguration c) throws IgniteException {
        IgniteConfiguration cfg = ctx.config();

        DeploymentMode depMode = cfg.getDeploymentMode();

        if (cfg.isPeerClassLoadingEnabled() && (depMode == PRIVATE || depMode == ISOLATED))
            throw new IgniteException("Cannot deploy services in PRIVATE or ISOLATED deployment mode: " + depMode);

        ensure(c.getName() != null, "getName() != null", null);
        ensure(c.getTotalCount() >= 0, "getTotalCount() >= 0", c.getTotalCount());
        ensure(c.getMaxPerNodeCount() >= 0, "getMaxPerNodeCount() >= 0", c.getMaxPerNodeCount());
        ensure(c.getService() != null, "getService() != null", c.getService());
        ensure(c.getTotalCount() > 0 || c.getMaxPerNodeCount() > 0,
            "c.getTotalCount() > 0 || c.getMaxPerNodeCount() > 0", null);
    }

    /**
     * @param cond Condition.
     * @param desc Description.
     * @param v Value.
     */
    private void ensure(boolean cond, String desc, @Nullable Object v) {
        if (!cond)
            if (v != null)
                throw new IgniteException("Service configuration check failed (" + desc + "): " + v);
            else
                throw new IgniteException("Service configuration check failed (" + desc + ")");
    }

    /**
     * @param name Service name.
     * @param svc Service.
     * @return Future.
     */
    public IgniteInternalFuture<?> deployNodeSingleton(ClusterGroup prj, String name, Service svc) {
        return deployMultiple(prj, name, svc, 0, 1);
    }

    /**
     * @param name Service name.
     * @param svc Service.
     * @return Future.
     */
    public IgniteInternalFuture<?> deployClusterSingleton(ClusterGroup prj, String name, Service svc) {
        return deployMultiple(prj, name, svc, 1, 1);
    }

    /**
     * @param name Service name.
     * @param svc Service.
     * @param totalCnt Total count.
     * @param maxPerNodeCnt Max per-node count.
     * @return Future.
     */
    public IgniteInternalFuture<?> deployMultiple(ClusterGroup prj, String name, Service svc, int totalCnt,
        int maxPerNodeCnt) {
        ServiceConfiguration cfg = new ServiceConfiguration();

        cfg.setName(name);
        cfg.setService(svc);
        cfg.setTotalCount(totalCnt);
        cfg.setMaxPerNodeCount(maxPerNodeCnt);

        return deployAll(prj, Collections.singleton(cfg));
    }

    /**
     * @param name Service name.
     * @param svc Service.
     * @param cacheName Cache name.
     * @param affKey Affinity key.
     * @return Future.
     */
    public IgniteInternalFuture<?> deployKeyAffinitySingleton(String name, Service svc, String cacheName,
        Object affKey) {
        A.notNull(affKey, "affKey");

        ServiceConfiguration cfg = new ServiceConfiguration();

        cfg.setName(name);
        cfg.setService(svc);
        cfg.setCacheName(cacheName);
        cfg.setAffinityKey(affKey);
        cfg.setTotalCount(1);
        cfg.setMaxPerNodeCount(1);

        // Ignore projection here.
        return deployAll(Collections.singleton(cfg), null);
    }

    /**
     * @param cfgs Service configurations.
     * @param dfltNodeFilter Default NodeFilter.
     * @return Configurations to deploy.
     */
    private PreparedConfigurations prepareServiceConfigurations(Collection<ServiceConfiguration> cfgs,
        IgnitePredicate<ClusterNode> dfltNodeFilter) {
        List<ServiceConfiguration> cfgsCp = new ArrayList<>(cfgs.size());

        Marshaller marsh = ctx.config().getMarshaller();

        List<GridServiceDeploymentFuture> failedFuts = null;

        for (ServiceConfiguration cfg : cfgs) {
            Exception err = null;

            // Deploy to projection node by default
            // or only on server nodes if no projection .
            if (cfg.getNodeFilter() == null && dfltNodeFilter != null)
                cfg.setNodeFilter(dfltNodeFilter);

            try {
                validate(cfg);
            }
            catch (Exception e) {
                U.error(log, "Failed to validate service configuration [name=" + cfg.getName() +
                    ", srvc=" + cfg.getService() + ']', e);

                err = e;
            }

            if (err == null) {
                try {
                    ctx.security().authorize(cfg.getName(), SecurityPermission.SERVICE_DEPLOY, null);
                }
                catch (Exception e) {
                    U.error(log, "Failed to authorize service creation [name=" + cfg.getName() +
                        ", srvc=" + cfg.getService() + ']', e);

                    err = e;
                }
            }

            if (err == null) {
                try {
                    byte[] srvcBytes = U.marshal(marsh, cfg.getService());

                    cfgsCp.add(new LazyServiceConfiguration(cfg, srvcBytes));
                }
                catch (Exception e) {
                    U.error(log, "Failed to marshal service with configured marshaller [name=" + cfg.getName() +
                        ", srvc=" + cfg.getService() + ", marsh=" + marsh + "]", e);

                    err = e;
                }
            }

            if (err != null) {
                if (failedFuts == null)
                    failedFuts = new ArrayList<>();

                GridServiceDeploymentFuture fut = new GridServiceDeploymentFuture(cfg);

                fut.onDone(err);

                failedFuts.add(fut);
            }
        }

        return new PreparedConfigurations(cfgsCp, failedFuts);
    }

    /**
     * @param prj Grid projection.
     * @param cfgs Service configurations.
     * @return Future for deployment.
     */
    public IgniteInternalFuture<?> deployAll(ClusterGroup prj, Collection<ServiceConfiguration> cfgs) {
        if (prj == null)
            // Deploy to servers by default if no projection specified.
            return deployAll(cfgs,  ctx.cluster().get().forServers().predicate());
        else if (prj.predicate() == F.<ClusterNode>alwaysTrue())
            return deployAll(cfgs,  null);
        else
            // Deploy to predicate nodes by default.
            return deployAll(cfgs,  prj.predicate());
    }

    /**
     * @param cfgs Service configurations.
     * @param dfltNodeFilter Default NodeFilter.
     * @return Future for deployment.
     */
    private IgniteInternalFuture<?> deployAll(Collection<ServiceConfiguration> cfgs,
        @Nullable IgnitePredicate<ClusterNode> dfltNodeFilter) {
        assert cfgs != null;

        PreparedConfigurations srvCfg = prepareServiceConfigurations(cfgs, dfltNodeFilter);

        List<ServiceConfiguration> cfgsCp = srvCfg.cfgs;

        List<GridServiceDeploymentFuture> failedFuts = srvCfg.failedFuts;

        Collections.sort(cfgsCp, new Comparator<ServiceConfiguration>() {
            @Override public int compare(ServiceConfiguration cfg1, ServiceConfiguration cfg2) {
                return cfg1.getName().compareTo(cfg2.getName());
            }
        });

        GridServiceDeploymentCompoundFuture res;

        while (true) {
            res = new GridServiceDeploymentCompoundFuture();

            if (ctx.deploy().enabled())
                ctx.cache().context().deploy().ignoreOwnership(true);

            try {
                if (cfgsCp.size() == 1)
                    writeServiceToCache(res, cfgsCp.get(0));
                else if (cfgsCp.size() > 1) {
                    try (Transaction tx = serviceCache().txStart(PESSIMISTIC, READ_COMMITTED)) {
                        for (ServiceConfiguration cfg : cfgsCp) {
                            try {
                                writeServiceToCache(res, cfg);
                            }
                            catch (IgniteCheckedException e) {
                                if (X.hasCause(e, ClusterTopologyCheckedException.class))
                                    throw e; // Retry.
                                else
                                    U.error(log, e.getMessage());
                            }
                        }

                        tx.commit();
                    }
                }

                break;
            }
            catch (IgniteException | IgniteCheckedException e) {
                for (String name : res.servicesToRollback())
                    depFuts.remove(name).onDone(e);

                if (X.hasCause(e, ClusterTopologyCheckedException.class)) {
                    if (log.isDebugEnabled())
                        log.debug("Topology changed while deploying services (will retry): " + e.getMessage());
                }
                else {
                    res.onDone(new IgniteCheckedException(
                        new ServiceDeploymentException("Failed to deploy provided services.", e, cfgs)));

                    return res;
                }
            }
            finally {
                if (ctx.deploy().enabled())
                    ctx.cache().context().deploy().ignoreOwnership(false);
            }
        }

        if (ctx.clientDisconnected()) {
            IgniteClientDisconnectedCheckedException err =
                new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
                    "Failed to deploy services, client node disconnected: " + cfgs);

            for (String name : res.servicesToRollback()) {
                GridServiceDeploymentFuture fut = depFuts.remove(name);

                if (fut != null)
                    fut.onDone(err);
            }

            return new GridFinishedFuture<>(err);
        }

        if (failedFuts != null) {
            for (GridServiceDeploymentFuture fut : failedFuts)
                res.add(fut, false);
        }

        res.markInitialized();

        return res;
    }

    /**
     * @param res Resulting compound future.
     * @param cfg Service configuration.
     * @throws IgniteCheckedException If operation failed.
     */
    private void writeServiceToCache(GridServiceDeploymentCompoundFuture res, ServiceConfiguration cfg)
        throws IgniteCheckedException {
        String name = cfg.getName();

        GridServiceDeploymentFuture fut = new GridServiceDeploymentFuture(cfg);

        GridServiceDeploymentFuture old = depFuts.putIfAbsent(name, fut);

        try {
            if (old != null) {
                if (!old.configuration().equalsIgnoreNodeFilter(cfg))
                    throw new IgniteCheckedException("Failed to deploy service (service already exists with different " +
                        "configuration) [deployed=" + old.configuration() + ", new=" + cfg + ']');
                else {
                    res.add(old, false);

                    return;
                }
            }

            GridServiceDeploymentKey key = new GridServiceDeploymentKey(name);

            GridServiceDeployment dep = (GridServiceDeployment)serviceCache().getAndPutIfAbsent(key,
                new GridServiceDeployment(ctx.localNodeId(), cfg));

            if (dep != null) {
                if (!dep.configuration().equalsIgnoreNodeFilter(cfg)) {
                    throw new IgniteCheckedException("Failed to deploy service (service already exists with " +
                        "different configuration) [deployed=" + dep.configuration() + ", new=" + cfg + ']');
                }
                else {
                    res.add(fut, false);

                    Iterator<Cache.Entry<Object, Object>> it = serviceEntries(ServiceAssignmentsPredicate.INSTANCE);

                    while (it.hasNext()) {
                        Cache.Entry<Object, Object> e = it.next();

                        GridServiceAssignments assigns = (GridServiceAssignments)e.getValue();

                        if (assigns.name().equals(name)) {
                            fut.onDone();

                            depFuts.remove(name, fut);

                            break;
                        }
                    }
                }
            }
            else
                res.add(fut, true);
        }
        catch (IgniteCheckedException e) {
            fut.onDone(e);

            res.add(fut, false);

            depFuts.remove(name, fut);

            throw e;
        }
    }

    /**
     * @param name Service name.
     * @return Future.
     */
    public IgniteInternalFuture<?> cancel(String name) {
        while (true) {
            try {
                return removeServiceFromCache(name).fut;
            }
            catch (IgniteException | IgniteCheckedException e) {
                if (X.hasCause(e, ClusterTopologyCheckedException.class)) {
                    if (log.isDebugEnabled())
                        log.debug("Topology changed while cancelling service (will retry): " + e.getMessage());
                }
                else {
                    U.error(log, "Failed to undeploy service: " + name, e);

                    return new GridFinishedFuture<>(e);
                }
            }
        }
    }

    /**
     * @return Future.
     */
    @SuppressWarnings("unchecked")
    public IgniteInternalFuture<?> cancelAll() {
        Iterator<Cache.Entry<Object, Object>> it = serviceEntries(ServiceDeploymentPredicate.INSTANCE);

        List<String> svcNames = new ArrayList<>();

        while (it.hasNext()) {
            GridServiceDeployment dep = (GridServiceDeployment)it.next().getValue();

            svcNames.add(dep.configuration().getName());
        }

        return cancelAll(svcNames);
    }

    /**
     * @param svcNames Name of service to deploy.
     * @return Future.
     */
    @SuppressWarnings("unchecked")
    public IgniteInternalFuture<?> cancelAll(Collection<String> svcNames) {
        List<String> svcNamesCp = new ArrayList<>(svcNames);

        Collections.sort(svcNamesCp);

        GridCompoundFuture res;

        while (true) {
            res = null;

            List<String> toRollback = new ArrayList<>();

            try (Transaction tx = serviceCache().txStart(PESSIMISTIC, READ_COMMITTED)) {
                for (String name : svcNames) {
                    if (res == null)
                        res = new GridCompoundFuture<>();

                    try {
                        CancelResult cr = removeServiceFromCache(name);

                        if (cr.rollback)
                            toRollback.add(name);

                        res.add(cr.fut);
                    }
                    catch (IgniteException | IgniteCheckedException e) {
                        if (X.hasCause(e, ClusterTopologyCheckedException.class))
                            throw e; // Retry.
                        else {
                            U.error(log, "Failed to undeploy service: " + name, e);

                            res.add(new GridFinishedFuture<>(e));
                        }
                    }
                }

                tx.commit();

                break;
            }
            catch (IgniteException | IgniteCheckedException e) {
                for (String name : toRollback)
                    undepFuts.remove(name).onDone(e);

                if (X.hasCause(e, ClusterTopologyCheckedException.class)) {
                    if (log.isDebugEnabled())
                        log.debug("Topology changed while cancelling service (will retry): " + e.getMessage());
                }
                else
                    return new GridFinishedFuture<>(e);
            }
        }

        if (res != null) {
            res.markInitialized();

            return res;
        }
        else
            return new GridFinishedFuture<>();
    }

    /**
     * @param name Name of service to remove from internal cache.
     * @return Cancellation future and a flag whether it should be completed and removed on error.
     * @throws IgniteCheckedException If operation failed.
     */
    private CancelResult removeServiceFromCache(String name) throws IgniteCheckedException {
        try {
            ctx.security().authorize(name, SecurityPermission.SERVICE_CANCEL, null);
        }
        catch (SecurityException e) {
            return new CancelResult(new GridFinishedFuture<>(e), false);
        }

        GridFutureAdapter<?> fut = new GridFutureAdapter<>();

        GridFutureAdapter<?> old = undepFuts.putIfAbsent(name, fut);

        if (old != null)
            return new CancelResult(old, false);
        else {
            GridServiceDeploymentKey key = new GridServiceDeploymentKey(name);

            try {
                if (serviceCache().getAndRemove(key) == null) {
                    // Remove future from local map if service was not deployed.
                    undepFuts.remove(name, fut);

                    fut.onDone();

                    return new CancelResult(fut, false);
                }
                else
                    return new CancelResult(fut, true);
            }
            catch (IgniteCheckedException e) {
                undepFuts.remove(name, fut);

                fut.onDone(e);

                throw e;
            }
        }
    }

    /**
     * @param name Service name.
     * @param timeout If greater than 0 limits task execution time. Cannot be negative.
     * @return Service topology.
     * @throws IgniteCheckedException On error.
     */
    public Map<UUID, Integer> serviceTopology(String name, long timeout) throws IgniteCheckedException {
        IgniteInternalCache<Object, Object> cache = serviceCache();

        ClusterNode node = cache.affinity().mapKeyToNode(name);

        final ServiceTopologyCallable call = new ServiceTopologyCallable(name);

        return ctx.closure().callAsyncNoFailover(
            GridClosureCallMode.BROADCAST,
            call,
            Collections.singletonList(node),
            false,
            timeout,
            true).get();
    }

    /**
     * @param cache Utility cache.
     * @param svcName Service name.
     * @return Service topology.
     * @throws IgniteCheckedException In case of error.
     */
    private static Map<UUID, Integer> serviceTopology(IgniteInternalCache<Object, Object> cache, String svcName)
        throws IgniteCheckedException {
        GridServiceAssignments val = (GridServiceAssignments)cache.get(new GridServiceAssignmentsKey(svcName));

        return val != null ? val.assigns() : null;
    }

    /**
     * @return Collection of service descriptors.
     */
    public Collection<ServiceDescriptor> serviceDescriptors() {
        Collection<ServiceDescriptor> descs = new ArrayList<>();

        Iterator<Cache.Entry<Object, Object>> it = serviceEntries(ServiceDeploymentPredicate.INSTANCE);

        while (it.hasNext()) {
            Cache.Entry<Object, Object> e = it.next();

            GridServiceDeployment dep = (GridServiceDeployment)e.getValue();

            ServiceDescriptorImpl desc = new ServiceDescriptorImpl(dep);

            try {
                GridServiceAssignments assigns = (GridServiceAssignments)serviceCache().getForcePrimary(
                    new GridServiceAssignmentsKey(dep.configuration().getName()));

                if (assigns != null) {
                    desc.topologySnapshot(assigns.assigns());

                    descs.add(desc);
                }
            }
            catch (IgniteCheckedException ex) {
                log.error("Failed to get assignments from replicated cache for service: " +
                    dep.configuration().getName(), ex);
            }
        }

        return descs;
    }

    /**
     * @param name Service name.
     * @param <T> Service type.
     * @return Service by specified service name.
     */
    @SuppressWarnings("unchecked")
    public <T> T service(String name) {
        ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null);

        Collection<ServiceContextImpl> ctxs;

        synchronized (locSvcs) {
            ctxs = locSvcs.get(name);
        }

        if (ctxs == null)
            return null;

        synchronized (ctxs) {
            if (ctxs.isEmpty())
                return null;

            for (ServiceContextImpl ctx : ctxs) {
                Service svc = ctx.service();

                if (svc != null)
                    return (T)svc;
            }

            return null;
        }
    }

    /**
     * @param name Service name.
     * @return Service by specified service name.
     */
    public ServiceContextImpl serviceContext(String name) {
        Collection<ServiceContextImpl> ctxs;

        synchronized (locSvcs) {
            ctxs = locSvcs.get(name);
        }

        if (ctxs == null)
            return null;

        synchronized (ctxs) {
            if (ctxs.isEmpty())
                return null;

            for (ServiceContextImpl ctx : ctxs) {
                if (ctx.service() != null)
                    return ctx;
            }

            return null;
        }
    }

    /**
     * @param prj Grid projection.
     * @param name Service name.
     * @param svcItf Service class.
     * @param sticky Whether multi-node request should be done.
     * @param timeout If greater than 0 limits service acquire time. Cannot be negative.
     * @param <T> Service interface type.
     * @return The proxy of a service by its name and class.
     * @throws IgniteException If failed to create proxy.
     */
    @SuppressWarnings("unchecked")
    public <T> T serviceProxy(ClusterGroup prj, String name, Class<? super T> svcItf, boolean sticky, long timeout)
        throws IgniteException {
        ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null);

        if (hasLocalNode(prj)) {
            ServiceContextImpl ctx = serviceContext(name);

            if (ctx != null) {
                Service svc = ctx.service();

                if (svc != null) {
                    if (!svcItf.isAssignableFrom(svc.getClass()))
                        throw new IgniteException("Service does not implement specified interface [svcItf=" +
                            svcItf.getName() + ", svcCls=" + svc.getClass().getName() + ']');

                    return (T)svc;
                }
            }
        }

        return new GridServiceProxy<T>(prj, name, svcItf, sticky, timeout, ctx).proxy();
    }

    /**
     * @param prj Grid nodes projection.
     * @return Whether given projection contains any local node.
     */
    private boolean hasLocalNode(ClusterGroup prj) {
        for (ClusterNode n : prj.nodes()) {
            if (n.isLocal())
                return true;
        }

        return false;
    }

    /**
     * @param name Service name.
     * @param <T> Service type.
     * @return Services by specified service name.
     */
    @SuppressWarnings("unchecked")
    public <T> Collection<T> services(String name) {
        ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null);

        Collection<ServiceContextImpl> ctxs;

        synchronized (locSvcs) {
            ctxs = locSvcs.get(name);
        }

        if (ctxs == null)
            return null;

        synchronized (ctxs) {
            Collection<T> res = new ArrayList<>(ctxs.size());

            for (ServiceContextImpl ctx : ctxs) {
                Service svc = ctx.service();

                if (svc != null)
                    res.add((T)svc);
            }

            return res;
        }
    }

    /**
     * Reassigns service to nodes.
     *
     * @param dep Service deployment.
     * @param topVer Topology version.
     * @throws IgniteCheckedException If failed.
     */
    private void reassign(GridServiceDeployment dep, AffinityTopologyVersion topVer) throws IgniteCheckedException {
        IgniteInternalCache<Object, Object> cache = serviceCache();

        ServiceConfiguration cfg = dep.configuration();
        Object nodeFilter = cfg.getNodeFilter();

        if (nodeFilter != null)
            ctx.resource().injectGeneric(nodeFilter);

        int totalCnt = cfg.getTotalCount();
        int maxPerNodeCnt = cfg.getMaxPerNodeCount();
        String cacheName = cfg.getCacheName();
        Object affKey = cfg.getAffinityKey();

        while (true) {
            GridServiceAssignments assigns = new GridServiceAssignments(cfg, dep.nodeId(), topVer.topologyVersion());

            Collection<ClusterNode> nodes;

            // Call node filter outside of transaction.
            if (affKey == null) {
                nodes = ctx.discovery().nodes(topVer);

                if (assigns.nodeFilter() != null) {
                    Collection<ClusterNode> nodes0 = new ArrayList<>();

                    for (ClusterNode node : nodes) {
                        if (assigns.nodeFilter().apply(node))
                            nodes0.add(node);
                    }

                    nodes = nodes0;
                }
            }
            else
                nodes = null;

            try (GridNearTxLocal tx = cache.txStartEx(PESSIMISTIC, REPEATABLE_READ)) {
                GridServiceAssignmentsKey key = new GridServiceAssignmentsKey(cfg.getName());

                GridServiceAssignments oldAssigns = (GridServiceAssignments)cache.get(key);

                Map<UUID, Integer> cnts = new HashMap<>();

                if (affKey != null) {
                    ClusterNode n = ctx.affinity().mapKeyToNode(cacheName, affKey, topVer);

                    if (n != null) {
                        int cnt = maxPerNodeCnt == 0 ? totalCnt == 0 ? 1 : totalCnt : maxPerNodeCnt;

                        cnts.put(n.id(), cnt);
                    }
                }
                else {
                    if (!nodes.isEmpty()) {
                        int size = nodes.size();

                        int perNodeCnt = totalCnt != 0 ? totalCnt / size : maxPerNodeCnt;
                        int remainder = totalCnt != 0 ? totalCnt % size : 0;

                        if (perNodeCnt >= maxPerNodeCnt && maxPerNodeCnt != 0) {
                            perNodeCnt = maxPerNodeCnt;
                            remainder = 0;
                        }

                        for (ClusterNode n : nodes)
                            cnts.put(n.id(), perNodeCnt);

                        assert perNodeCnt >= 0;
                        assert remainder >= 0;

                        if (remainder > 0) {
                            int cnt = perNodeCnt + 1;

                            if (oldAssigns != null) {
                                Collection<UUID> used = new HashSet<>();

                                // Avoid redundant moving of services.
                                for (Map.Entry<UUID, Integer> e : oldAssigns.assigns().entrySet()) {
                                    // Do not assign services to left nodes.
                                    if (ctx.discovery().node(e.getKey()) == null)
                                        continue;

                                    // If old count and new count match, then reuse the assignment.
                                    if (e.getValue() == cnt) {
                                        cnts.put(e.getKey(), cnt);

                                        used.add(e.getKey());

                                        if (--remainder == 0)
                                            break;
                                    }
                                }

                                if (remainder > 0) {
                                    List<Map.Entry<UUID, Integer>> entries = new ArrayList<>(cnts.entrySet());

                                    // Randomize.
                                    Collections.shuffle(entries);

                                    for (Map.Entry<UUID, Integer> e : entries) {
                                        // Assign only the ones that have not been reused from previous assignments.
                                        if (!used.contains(e.getKey())) {
                                            if (e.getValue() < maxPerNodeCnt || maxPerNodeCnt == 0) {
                                                e.setValue(e.getValue() + 1);

                                                if (--remainder == 0)
                                                    break;
                                            }
                                        }
                                    }
                                }
                            }
                            else {
                                List<Map.Entry<UUID, Integer>> entries = new ArrayList<>(cnts.entrySet());

                                // Randomize.
                                Collections.shuffle(entries);

                                for (Map.Entry<UUID, Integer> e : entries) {
                                    e.setValue(e.getValue() + 1);

                                    if (--remainder == 0)
                                        break;
                                }
                            }
                        }
                    }
                }

                assigns.assigns(cnts);

                cache.put(key, assigns);

                tx.commit();

                break;
            }
            catch (ClusterTopologyCheckedException e) {
                if (log.isDebugEnabled())
                    log.debug("Topology changed while reassigning (will retry): " + e.getMessage());

                U.sleep(10);
            }
        }
    }

    /**
     * Redeploys local services based on assignments.
     *
     * @param assigns Assignments.
     */
    private void redeploy(GridServiceAssignments assigns) {
        if (assigns.topologyVersion() < ctx.discovery().topologyVersion()) {
            if (log.isDebugEnabled())
                log.debug("Skip outdated assignment [assigns=" + assigns +
                    ", topVer=" + ctx.discovery().topologyVersion() + ']');

            return;
        }

        String svcName = assigns.name();

        Integer assignCnt = assigns.assigns().get(ctx.localNodeId());

        if (assignCnt == null)
            assignCnt = 0;

        Collection<ServiceContextImpl> ctxs;

        synchronized (locSvcs) {
            ctxs = locSvcs.get(svcName);

            if (ctxs == null)
                locSvcs.put(svcName, ctxs = new ArrayList<>());
        }

        Collection<ServiceContextImpl> toInit = new ArrayList<>();

        synchronized (ctxs) {
            if (ctxs.size() > assignCnt) {
                int cancelCnt = ctxs.size() - assignCnt;

                cancel(ctxs, cancelCnt);
            }
            else if (ctxs.size() < assignCnt) {
                int createCnt = assignCnt - ctxs.size();

                for (int i = 0; i < createCnt; i++) {
                    ServiceContextImpl svcCtx = new ServiceContextImpl(assigns.name(),
                        UUID.randomUUID(),
                        assigns.cacheName(),
                        assigns.affinityKey(),
                        Executors.newSingleThreadExecutor(threadFactory));

                    ctxs.add(svcCtx);

                    toInit.add(svcCtx);
                }
            }
        }

        for (final ServiceContextImpl svcCtx : toInit) {
            final Service svc;

            try {
                svc = copyAndInject(assigns.configuration());

                // Initialize service.
                svc.init(svcCtx);

                svcCtx.service(svc);
            }
            catch (Throwable e) {
                U.error(log, "Failed to initialize service (service will not be deployed): " + assigns.name(), e);

                synchronized (ctxs) {
                    ctxs.removeAll(toInit);
                }

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

                if (e instanceof RuntimeException)
                    throw (RuntimeException)e;

                return;
            }

            if (log.isInfoEnabled())
                log.info("Starting service instance [name=" + svcCtx.name() + ", execId=" +
                    svcCtx.executionId() + ']');

            // Start service in its own thread.
            final ExecutorService exe = svcCtx.executor();

            exe.execute(new Runnable() {
                @Override public void run() {
                    try {
                        svc.execute(svcCtx);
                    }
                    catch (InterruptedException | IgniteInterruptedCheckedException ignore) {
                        if (log.isDebugEnabled())
                            log.debug("Service thread was interrupted [name=" + svcCtx.name() + ", execId=" +
                                svcCtx.executionId() + ']');
                    }
                    catch (IgniteException e) {
                        if (e.hasCause(InterruptedException.class) ||
                            e.hasCause(IgniteInterruptedCheckedException.class)) {
                            if (log.isDebugEnabled())
                                log.debug("Service thread was interrupted [name=" + svcCtx.name() +
                                    ", execId=" + svcCtx.executionId() + ']');
                        }
                        else {
                            U.error(log, "Service execution stopped with error [name=" + svcCtx.name() +
                                ", execId=" + svcCtx.executionId() + ']', e);
                        }
                    }
                    catch (Throwable e) {
                        log.error("Service execution stopped with error [name=" + svcCtx.name() +
                            ", execId=" + svcCtx.executionId() + ']', e);

                        if (e instanceof Error)
                            throw (Error)e;
                    }
                    finally {
                        // Suicide.
                        exe.shutdownNow();
                    }
                }
            });
        }
    }

    /**
     * @param cfg Service configuration.
     * @return Copy of service.
     * @throws IgniteCheckedException If failed.
     */
    private Service copyAndInject(ServiceConfiguration cfg) throws IgniteCheckedException {
        Marshaller m = ctx.config().getMarshaller();

        if (cfg instanceof LazyServiceConfiguration) {
            byte[] bytes = ((LazyServiceConfiguration)cfg).serviceBytes();

            Service srvc = U.unmarshal(m, bytes, U.resolveClassLoader(null, ctx.config()));

            ctx.resource().inject(srvc);

            return srvc;
        }
        else {
            Service svc = cfg.getService();

            try {
                byte[] bytes = U.marshal(m, svc);

                Service cp = U.unmarshal(m, bytes, U.resolveClassLoader(svc.getClass().getClassLoader(), ctx.config()));

                ctx.resource().inject(cp);

                return cp;
            }
            catch (IgniteCheckedException e) {
                U.error(log, "Failed to copy service (will reuse same instance): " + svc.getClass(), e);

                return svc;
            }
        }
    }

    /**
     * @param ctxs Contexts to cancel.
     * @param cancelCnt Number of contexts to cancel.
     */
    private void cancel(Iterable<ServiceContextImpl> ctxs, int cancelCnt) {
        for (Iterator<ServiceContextImpl> it = ctxs.iterator(); it.hasNext(); ) {
            ServiceContextImpl svcCtx = it.next();

            // Flip cancelled flag.
            svcCtx.setCancelled(true);

            // Notify service about cancellation.
            Service svc = svcCtx.service();

            if (svc != null) {
                try {
                    svc.cancel(svcCtx);
                }
                catch (Throwable e) {
                    log.error("Failed to cancel service (ignoring) [name=" + svcCtx.name() +
                        ", execId=" + svcCtx.executionId() + ']', e);

                    if (e instanceof Error)
                        throw e;
                }
                finally {
                    try {
                        ctx.resource().cleanup(svc);
                    }
                    catch (IgniteCheckedException e) {
                        U.error(log, "Failed to clean up service (will ignore): " + svcCtx.name(), e);
                    }
                }
            }

            // Close out executor thread for the service.
            // This will cause the thread to be interrupted.
            svcCtx.executor().shutdownNow();

            it.remove();

            if (log.isInfoEnabled())
                log.info("Cancelled service instance [name=" + svcCtx.name() + ", execId=" +
                    svcCtx.executionId() + ']');

            if (--cancelCnt == 0)
                break;
        }
    }

    /**
     * @param p Entry predicate used to execute query from client node.
     * @return Service deployment entries.
     */
    @SuppressWarnings("unchecked")
    private Iterator<Cache.Entry<Object, Object>> serviceEntries(IgniteBiPredicate<Object, Object> p) {
        try {
            IgniteInternalCache<Object, Object> cache = serviceCache();

            GridCacheQueryManager qryMgr = cache.context().queries();

            CacheQuery<Map.Entry<Object, Object>> qry = qryMgr.createScanQuery(p, null, false);

            DiscoveryDataClusterState clusterState = ctx.state().clusterState();

            if ((clusterState.hasBaselineTopology()
                && !CU.baselineNode(ctx.cluster().get().localNode(), clusterState))
                || !cache.context().affinityNode()) {
                ClusterNode oldestSrvNode =
                    ctx.discovery().oldestAliveServerNode(AffinityTopologyVersion.NONE);

                if (oldestSrvNode == null)
                    return new GridEmptyIterator<>();

                qry.projection(ctx.cluster().get().forNode(oldestSrvNode));
            }
            else
                qry.projection(ctx.cluster().get().forLocal());

            GridCloseableIterator<Map.Entry<Object, Object>> iter = qry.executeScanQuery();

            return cache.context().itHolder().iterator(iter,
                new CacheIteratorConverter<Cache.Entry<Object, Object>, Map.Entry<Object, Object>>() {
                    @Override protected Cache.Entry<Object, Object> convert(Map.Entry<Object, Object> e) {
                        // Actually Scan Query returns Iterator<CacheQueryEntry> by default,
                        // CacheQueryEntry implements both Map.Entry and Cache.Entry interfaces.
                        return (Cache.Entry<Object, Object>)e;
                    }

                    @Override protected void remove(Cache.Entry<Object, Object> item) {
                        throw new UnsupportedOperationException();
                    }
                });
        }
        catch (IgniteCheckedException e) {
            throw new IgniteException(e);
        }
    }

    /**
     * Called right after utility cache is started and ready for the usage.
     */
    public void onUtilityCacheStarted() {
        synchronized (pendingJobCtxs) {
            if (pendingJobCtxs.isEmpty())
                return;

            Iterator<ComputeJobContext> iter = pendingJobCtxs.iterator();

            while (iter.hasNext()) {
                iter.next().callcc();
                iter.remove();
            }
        }
    }

    /**
     * Service deployment listener.
     */
    @SuppressWarnings("unchecked")
    private class ServiceEntriesListener implements CacheEntryUpdatedListener<Object, Object> {
        /** {@inheritDoc} */
        @Override public void onUpdated(final Iterable<CacheEntryEvent<?, ?>> deps) {
            GridSpinBusyLock busyLock = GridServiceProcessor.this.busyLock;

            if (busyLock == null || !busyLock.enterBusy())
                return;

            try {
                depExe.execute(new DepRunnable() {
                    @Override public void run0() {
                        onSystemCacheUpdated(deps);
                    }
                });
            }
            finally {
                busyLock.leaveBusy();
            }
        }
    }

    /**
     * @param evts Update events.
     */
    private void onSystemCacheUpdated(final Iterable<CacheEntryEvent<?, ?>> evts) {
        for (CacheEntryEvent<?, ?> e : evts) {
            if (e.getKey() instanceof GridServiceDeploymentKey)
                processDeployment((CacheEntryEvent)e);
            else if (e.getKey() instanceof GridServiceAssignmentsKey)
                processAssignment((CacheEntryEvent)e);
        }
    }

    /**
     * @param e Entry.
     */
    private void processDeployment(CacheEntryEvent<GridServiceDeploymentKey, GridServiceDeployment> e) {
        GridServiceDeployment dep;

        try {
            dep = e.getValue();
        }
        catch (IgniteException ex) {
            if (X.hasCause(ex, ClassNotFoundException.class))
                return;
            else
                throw ex;
        }

        if (e.getEventType() != REMOVED) {
            svcName.set(dep.configuration().getName());

            // Ignore other utility cache events.
            AffinityTopologyVersion topVer = ctx.discovery().topologyVersionEx();

            ClusterNode oldest = U.oldest(ctx.discovery().nodes(topVer), null);

            // Process deployment on coordinator only.
            if (oldest.isLocal())
                onDeployment(dep, topVer);
        }
        // Handle undeployment.
        else {
            String name = e.getKey().name();

            undeploy(name);

            // Finish deployment futures if undeployment happened.
            GridFutureAdapter<?> fut = depFuts.remove(name);

            if (fut != null)
                fut.onDone();

            // Complete undeployment future.
            fut = undepFuts.remove(name);

            if (fut != null)
                fut.onDone();

            GridServiceAssignmentsKey key = new GridServiceAssignmentsKey(name);

            // Remove assignment on primary node in case of undeploy.
            IgniteInternalCache<Object, Object> cache = serviceCache();

            if (cache.cache().affinity().isPrimary(ctx.discovery().localNode(), key)) {
                try {
                    cache.getAndRemove(key);
                }
                catch (IgniteCheckedException ex) {
                    U.error(log, "Failed to remove assignments for undeployed service: " + name, ex);
                }
            }
        }
    }

    /**
     * Deployment callback.
     *
     * @param dep Service deployment.
     * @param topVer Topology version.
     */
    private void onDeployment(final GridServiceDeployment dep, final AffinityTopologyVersion topVer) {
        // Retry forever.
        try {
            AffinityTopologyVersion newTopVer = ctx.discovery().topologyVersionEx();

            // If topology version changed, reassignment will happen from topology event.
            if (newTopVer.equals(topVer))
                reassign(dep, topVer);
        }
        catch (IgniteCheckedException e) {
            if (!(e instanceof ClusterTopologyCheckedException))
                log.error("Failed to do service reassignment (will retry): " + dep.configuration().getName(), e);

            AffinityTopologyVersion newTopVer = ctx.discovery().topologyVersionEx();

            if (!newTopVer.equals(topVer)) {
                assert newTopVer.compareTo(topVer) > 0;

                // Reassignment will happen from topology event.
                return;
            }

            ctx.timeout().addTimeoutObject(new GridTimeoutObject() {
                private IgniteUuid id = IgniteUuid.randomUuid();

                private long start = System.currentTimeMillis();

                @Override public IgniteUuid timeoutId() {
                    return id;
                }

                @Override public long endTime() {
                    return start + RETRY_TIMEOUT;
                }

                @Override public void onTimeout() {
                    depExe.execute(new DepRunnable() {
                        @Override public void run0() {
                            onDeployment(dep, topVer);
                        }
                    });
                }
            });
        }
    }

    /**
     * Topology listener.
     */
    private class TopologyListener implements DiscoveryEventListener {
        /** */
        private volatile AffinityTopologyVersion currTopVer = null;

        /**
         * Check that listening-in topology version is the latest and wait until exchange is finished.
         *
         * @param initTopVer listening-in topology version.
         * @return {@code True} if current event is not last and should be skipped.
         */
        private boolean skipExchange(final AffinityTopologyVersion initTopVer) {
            AffinityTopologyVersion pendingTopVer = null;
            AffinityTopologyVersion newTopVer;

            if (!initTopVer.equals(newTopVer = currTopVer))
                pendingTopVer = newTopVer;
            else {
                IgniteInternalFuture<?> affReadyFut = ctx.cache().context().exchange().affinityReadyFuture(initTopVer);

                if (affReadyFut != null) {
                    try {
                        affReadyFut.get();
                    }
                    catch (IgniteCheckedException e) {
                        U.warn(log, "Failed to wait for affinity ready future " +
                            "(the assignment will be recalculated anyway):" + e.toString());
                    }
                }

                // If exchange already moved forward - skip current version.
                if (!initTopVer.equals(newTopVer = currTopVer))
                    pendingTopVer = newTopVer;
            }

            boolean skipExchange = pendingTopVer != null;

            if (skipExchange && log.isInfoEnabled()) {
                log.info("Service processor detected a topology change during " +
                    "assignments calculation (will abort current iteration and " +
                    "re-calculate on the newer version): " +
                    "[topVer=" + initTopVer + ", newTopVer=" + pendingTopVer + ']');
            }

            return skipExchange;
        }

        /** {@inheritDoc} */
        @Override public void onEvent(final DiscoveryEvent evt, final DiscoCache discoCache) {
            GridSpinBusyLock busyLock = GridServiceProcessor.this.busyLock;

            if (busyLock == null || !busyLock.enterBusy())
                return;

            try {
                final AffinityTopologyVersion topVer;

                if (evt instanceof DiscoveryCustomEvent) {
                    DiscoveryCustomMessage msg = ((DiscoveryCustomEvent)evt).customMessage();

                    if (msg instanceof CacheAffinityChangeMessage) {
                        if (!((CacheAffinityChangeMessage)msg).exchangeNeeded())
                            return;
                    }
                    else if (msg instanceof DynamicCacheChangeBatch) {
                        if (!((DynamicCacheChangeBatch)msg).exchangeNeeded())
                            return;
                    }
                    else
                        return;

                    if (msg instanceof MetadataUpdateProposedMessage || msg instanceof MetadataUpdateAcceptedMessage)
                        return;

                    topVer = ((DiscoveryCustomEvent)evt).affinityTopologyVersion();
                }
                else
                    topVer = new AffinityTopologyVersion((evt).topologyVersion(), 0);

                currTopVer = topVer;

                depExe.execute(new DepRunnable() {
                    @Override public void run0() {
                        // In case the cache instance isn't tracked by DiscoveryManager anymore.
                        discoCache.updateAlives(ctx.discovery());

                        ClusterNode oldest = discoCache.oldestAliveServerNode();

                        if (oldest != null && oldest.isLocal()) {
                            final Collection<GridServiceDeployment> retries = new ConcurrentLinkedQueue<>();

                            if (ctx.deploy().enabled())
                                ctx.cache().context().deploy().ignoreOwnership(true);

                            try {
                                Iterator<Cache.Entry<Object, Object>> it = serviceEntries(
                                    ServiceDeploymentPredicate.INSTANCE);

                                while (it.hasNext()) {
                                    // If topology changed again, let next event handle it.
                                    if (skipExchange(topVer))
                                        return;

                                    Cache.Entry<Object, Object> e = it.next();

                                    GridServiceDeployment dep = (GridServiceDeployment)e.getValue();

                                    try {
                                        svcName.set(dep.configuration().getName());

                                        reassign(dep, topVer);
                                    }
                                    catch (IgniteCheckedException ex) {
                                        if (!(ex instanceof ClusterTopologyCheckedException))
                                            LT.error(log, ex, "Failed to do service reassignment (will retry): " +
                                                dep.configuration().getName());

                                        retries.add(dep);
                                    }
                                }
                            }
                            finally {
                                if (ctx.deploy().enabled())
                                    ctx.cache().context().deploy().ignoreOwnership(false);
                            }

                            if (!retries.isEmpty())
                                onReassignmentFailed(topVer, retries);
                        }

                        Iterator<Cache.Entry<Object, Object>> it = serviceEntries(ServiceAssignmentsPredicate.INSTANCE);

                        // Clean up zombie assignments.
                        IgniteInternalCache<Object, Object> cache = serviceCache();

                        while (it.hasNext()) {
                            // If topology changed again, let next event handle it.
                            if (skipExchange(topVer))
                                return;

                            Cache.Entry<Object, Object> e = it.next();

                            if (cache.context().affinity().primaryByKey(ctx.grid().localNode(), e.getKey(), topVer)) {
                                String name = ((GridServiceAssignmentsKey)e.getKey()).name();

                                try {
                                    if (cache.get(new GridServiceDeploymentKey(name)) == null) {
                                        if (log.isDebugEnabled())
                                            log.debug("Removed zombie assignments: " + e.getValue());

                                        cache.getAndRemove(e.getKey());
                                    }
                                }
                                catch (IgniteCheckedException ex) {
                                    U.error(log, "Failed to clean up zombie assignments for service: " + name, ex);
                                }
                            }
                        }
                    }
                });
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /**
         * Handler for reassignment failures.
         *
         * @param topVer Topology version.
         * @param retries Retries.
         */
        private void onReassignmentFailed(final AffinityTopologyVersion topVer,
            final Collection<GridServiceDeployment> retries) {
            GridSpinBusyLock busyLock = GridServiceProcessor.this.busyLock;

            if (busyLock == null || !busyLock.enterBusy())
                return;

            try {
                // If topology changed again, let next event handle it.
                if (ctx.discovery().topologyVersionEx().equals(topVer))
                    return;

                for (Iterator<GridServiceDeployment> it = retries.iterator(); it.hasNext(); ) {
                    GridServiceDeployment dep = it.next();

                    try {
                        svcName.set(dep.configuration().getName());

                        reassign(dep, topVer);

                        it.remove();
                    }
                    catch (IgniteCheckedException e) {
                        if (!(e instanceof ClusterTopologyCheckedException))
                            LT.error(log, e, "Failed to do service reassignment (will retry): " +
                                dep.configuration().getName());
                    }
                }

                if (!retries.isEmpty()) {
                    ctx.timeout().addTimeoutObject(new GridTimeoutObject() {
                        private IgniteUuid id = IgniteUuid.randomUuid();

                        private long start = System.currentTimeMillis();

                        @Override public IgniteUuid timeoutId() {
                            return id;
                        }

                        @Override public long endTime() {
                            return start + RETRY_TIMEOUT;
                        }

                        @Override public void onTimeout() {
                            depExe.execute(new Runnable() {
                                @Override public void run() {
                                    onReassignmentFailed(topVer, retries);
                                }
                            });
                        }
                    });
                }
            }
            finally {
                busyLock.leaveBusy();
            }
        }
    }

    /**
     * @param e Entry.
     */
    private void processAssignment(CacheEntryEvent<GridServiceAssignmentsKey, GridServiceAssignments> e) {
        GridServiceAssignments assigns;

        try {
            assigns = e.getValue();
        }
        catch (IgniteException ex) {
            if (X.hasCause(ex, ClassNotFoundException.class))
                return;
            else
                throw ex;
        }

        if (e.getEventType() != REMOVED) {
            svcName.set(assigns.name());

            Throwable t = null;

            try {
                redeploy(assigns);
            }
            catch (Error | RuntimeException th) {
                t = th;
            }

            GridServiceDeploymentFuture fut = depFuts.get(assigns.name());

            if (fut != null && fut.configuration().equalsIgnoreNodeFilter(assigns.configuration())) {
                depFuts.remove(assigns.name(), fut);

                // Complete deployment futures once the assignments have been stored in cache.
                fut.onDone(null, t);
            }
        }
        // Handle undeployment.
        else
            undeploy(e.getKey().name());
    }

    /**
     * @param name Name.
     */
    private void undeploy(String name) {
        svcName.set(name);

        Collection<ServiceContextImpl> ctxs;

        synchronized (locSvcs) {
            ctxs = locSvcs.remove(name);
        }

        if (ctxs != null) {
            synchronized (ctxs) {
                cancel(ctxs, ctxs.size());
            }
        }
    }

    /**
     *
     */
    private static class CancelResult {
        /** */
        IgniteInternalFuture<?> fut;

        /** */
        boolean rollback;

        /**
         * @param fut Future.
         * @param rollback {@code True} if service was cancelled during current call.
         */
        CancelResult(IgniteInternalFuture<?> fut, boolean rollback) {
            this.fut = fut;
            this.rollback = rollback;
        }
    }

    /**
     *
     */
    private abstract class DepRunnable implements Runnable {
        /** {@inheritDoc} */
        @Override public void run() {
            GridSpinBusyLock busyLock = GridServiceProcessor.this.busyLock;

            if (busyLock == null || !busyLock.enterBusy())
                return;

            // Won't block ServiceProcessor stopping process.
            busyLock.leaveBusy();

            svcName.set(null);

            try {
                run0();
            }
            catch (Throwable t) {
                log.error("Error when executing service: " + svcName.get(), t);

                if (t instanceof Error)
                    throw t;
            }
            finally {
                svcName.set(null);
            }
        }

        /**
         * Abstract run method protected by busy lock.
         */
        public abstract void run0();
    }

    /**
     *
     */
    static class ServiceDeploymentPredicate implements IgniteBiPredicate<Object, Object> {
        /** */
        static final ServiceDeploymentPredicate INSTANCE = new ServiceDeploymentPredicate();

        /** */
        private static final long serialVersionUID = 0L;

        /** {@inheritDoc} */
        @Override public boolean apply(Object key, Object val) {
            return key instanceof GridServiceDeploymentKey;
        }

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

    /**
     *
     */
    static class ServiceAssignmentsPredicate implements IgniteBiPredicate<Object, Object> {
        /** */
        static final ServiceAssignmentsPredicate INSTANCE = new ServiceAssignmentsPredicate();

        /** */
        private static final long serialVersionUID = 0L;

        /** {@inheritDoc} */
        @Override public boolean apply(Object key, Object val) {
            return key instanceof GridServiceAssignmentsKey;
        }

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

    /**
     */
    @GridInternal
    private static class ServiceTopologyCallable implements IgniteCallable<Map<UUID, Integer>> {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private final String svcName;

        /** */
        private transient boolean waitedCacheInit;

        /** */
        @IgniteInstanceResource
        private IgniteEx ignite;

        /** */
        @JobContextResource
        private transient ComputeJobContext jCtx;

        /** */
        @LoggerResource
        private transient IgniteLogger log;

        /**
         * @param svcName Service name.
         */
        public ServiceTopologyCallable(String svcName) {
            this.svcName = svcName;
        }

        /** {@inheritDoc} */
        @Override public Map<UUID, Integer> call() throws Exception {
            IgniteInternalCache<Object, Object> cache = ignite.context().cache().utilityCache();

            if (cache == null) {
                List<ComputeJobContext> pendingCtxs = ignite.context().service().pendingJobCtxs;

                synchronized (pendingCtxs) {
                    // Double check cache reference after lock acqusition.
                    cache = ignite.context().cache().utilityCache();

                    if (cache == null) {
                        if (!waitedCacheInit) {
                            log.debug("Utility cache hasn't been initialized yet. Waiting.");

                            // waiting for a minute for cache initialization.
                            jCtx.holdcc(60 * 1000);

                            pendingCtxs.add(jCtx);

                            waitedCacheInit = true;

                            return null;
                        }
                        else {
                            log.error("Failed to gather service topology. Utility " +
                                "cache initialization is stuck.");

                            throw new IgniteCheckedException("Failed to gather service topology. Utility " +
                                "cache initialization is stuck.");
                        }
                    }
                }
            }

            return serviceTopology(cache, svcName);
        }
    }
}