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

import java.io.InputStream;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeoutException;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.configuration.DeploymentMode;
import org.apache.ignite.events.DeploymentEvent;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.util.GridAnnotationsCache;
import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashSet;
import org.apache.ignite.internal.util.GridClassLoaderCache;
import org.apache.ignite.internal.util.GridStripedLock;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.P1;
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.IgniteBiTuple;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.marshaller.AbstractMarshaller;
import org.apache.ignite.spi.deployment.DeploymentSpi;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.configuration.DeploymentMode.CONTINUOUS;
import static org.apache.ignite.configuration.DeploymentMode.SHARED;
import static org.apache.ignite.events.EventType.EVT_CLASS_DEPLOYED;
import static org.apache.ignite.events.EventType.EVT_CLASS_UNDEPLOYED;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static org.apache.ignite.events.EventType.EVT_TASK_DEPLOYED;
import static org.apache.ignite.events.EventType.EVT_TASK_UNDEPLOYED;

/**
 * Deployment storage for {@link org.apache.ignite.configuration.DeploymentMode#SHARED} and
 * {@link org.apache.ignite.configuration.DeploymentMode#CONTINUOUS} modes.
 */
public class GridDeploymentPerVersionStore extends GridDeploymentStoreAdapter {
    /** Shared deployment cache. */
    private final Map<String, List<SharedDeployment>> cache = new HashMap<>();

    /** Set of obsolete class loaders. */
    private final Collection<IgniteUuid> deadClsLdrs = new GridBoundedConcurrentLinkedHashSet<>(1024, 64);

    /** Discovery listener. */
    private GridLocalEventListener discoLsnr;

    /**
     * Resources hits/misses cache (by loader ID). If missed resources cache is turned off,
     * only hits are cached.
     */
    private final ConcurrentMap<IgniteUuid, Map<String, Boolean>> rsrcCache;

    /** Mutex. */
    private final Object mux = new Object();

    /** Missed resources cache size. */
    private final int missedRsrcCacheSize;

    /** Lock to synchronize remote load checks. */
    private final GridStripedLock loadRmtLock = new GridStripedLock(16);

    /**
     * @param spi Underlying SPI.
     * @param ctx Grid kernal context.
     * @param comm Deployment communication.
     */
    GridDeploymentPerVersionStore(DeploymentSpi spi, GridKernalContext ctx, GridDeploymentCommunication comm) {
        super(spi, ctx, comm);

        missedRsrcCacheSize = ctx.config().getPeerClassLoadingMissedResourcesCacheSize();

        rsrcCache = new ConcurrentHashMap<>();
    }

    /** {@inheritDoc} */
    @Override public void start() throws IgniteCheckedException {
        discoLsnr = new GridLocalEventListener() {
            @Override public void onEvent(Event evt) {
                assert evt instanceof DiscoveryEvent;

                assert evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED;

                DiscoveryEvent discoEvt = (DiscoveryEvent)evt;

                Collection<SharedDeployment> undeployed = new LinkedList<>();

                if (log.isDebugEnabled())
                    log.debug("Processing node departure event: " + evt);

                synchronized (mux) {
                    for (Iterator<List<SharedDeployment>> i1 = cache.values().iterator(); i1.hasNext();) {
                        List<SharedDeployment> deps = i1.next();

                        for (Iterator<SharedDeployment> i2 = deps.iterator(); i2.hasNext();) {
                            SharedDeployment dep = i2.next();

                            dep.removeParticipant(discoEvt.eventNode().id());

                            if (!dep.hasParticipants()) {
                                if (dep.deployMode() == SHARED) {
                                    if (!dep.undeployed()) {
                                        dep.undeploy();

                                        // Undeploy.
                                        i2.remove();

                                        assert !dep.isRemoved();

                                        dep.onRemoved();

                                        undeployed.add(dep);

                                        if (log.isDebugEnabled())
                                            log.debug("Undeployed class loader as there are no participating " +
                                                "nodes: " + dep);
                                    }
                                }
                                else {
                                    if (log.isDebugEnabled())
                                        log.debug("Preserving deployment without node participants: " + dep);
                                }
                            }
                            else {
                                if (log.isDebugEnabled())
                                    log.debug("Keeping deployment as it still has participants: " + dep);
                            }
                        }

                        if (deps.isEmpty())
                            i1.remove();
                    }
                }

                recordUndeployed(discoEvt.eventNode().id(), undeployed);
            }
        };

        ctx.event().addLocalEventListener(discoLsnr, EVT_NODE_FAILED, EVT_NODE_LEFT);

        if (log.isDebugEnabled())
            log.debug(startInfo());
    }

    /** {@inheritDoc} */
    @Override public void stop() {
        Collection<SharedDeployment> cp = new HashSet<>();

        synchronized (mux) {
            for (List<SharedDeployment> deps : cache.values())
                for (SharedDeployment dep : deps) {
                    // Mark undeployed.
                    dep.undeploy();

                    cp.add(dep);
                }

            cache.clear();
        }

        for (SharedDeployment dep : cp)
            dep.recordUndeployed(null);

        if (log.isDebugEnabled())
            log.debug(stopInfo());
    }

    /** {@inheritDoc} */
    @Override public void onKernalStart() throws IgniteCheckedException {
        Collection<SharedDeployment> undeployed = new LinkedList<>();

        synchronized (mux) {
            for (Iterator<List<SharedDeployment>> i1 = cache.values().iterator(); i1.hasNext();) {
                List<SharedDeployment> deps = i1.next();

                for (Iterator<SharedDeployment> i2 = deps.iterator(); i2.hasNext();) {
                    SharedDeployment dep = i2.next();

                    for (UUID nodeId : dep.getParticipantNodeIds())
                        if (ctx.discovery().node(nodeId) == null)
                            dep.removeParticipant(nodeId);

                    if (!dep.hasParticipants()) {
                        if (dep.deployMode() == SHARED) {
                            if (!dep.undeployed()) {
                                dep.undeploy();

                                // Undeploy.
                                i2.remove();

                                dep.onRemoved();

                                undeployed.add(dep);

                                if (log.isDebugEnabled())
                                    log.debug("Undeployed class loader as there are no participating nodes: " + dep);
                            }
                        }
                        else if (log.isDebugEnabled())
                            log.debug("Preserving deployment without node participants: " + dep);
                    }
                    else if (log.isDebugEnabled())
                        log.debug("Keeping deployment as it still has participants: " + dep);
                }

                if (deps.isEmpty())
                    i1.remove();
            }
        }

        recordUndeployed(null, undeployed);

        if (log.isDebugEnabled())
            log.debug("Registered deployment discovery listener: " + discoLsnr);
    }

    /** {@inheritDoc} */
    @Override public void onKernalStop() {
        if (discoLsnr != null) {
            ctx.event().removeLocalEventListener(discoLsnr);

            if (log.isDebugEnabled())
                log.debug("Unregistered deployment discovery listener: " + discoLsnr);
        }
    }

    /** {@inheritDoc} */
    @Override public Collection<GridDeployment> getDeployments() {
        Collection<GridDeployment> deps = new LinkedList<>();

        synchronized (mux) {
            for (List<SharedDeployment> list : cache.values())
                for (SharedDeployment d : list)
                    deps.add(d);
        }

        return deps;
    }

    /** {@inheritDoc} */
    @Override public GridDeployment getDeployment(final IgniteUuid ldrId) {
        synchronized (mux) {
            return F.find(F.flat(cache.values()), null, new P1<SharedDeployment>() {
                @Override public boolean apply(SharedDeployment d) {
                    return d.classLoaderId().equals(ldrId);
                }
            });
        }
    }

    /** {@inheritDoc} */
    @Override public GridDeployment searchDeploymentCache(GridDeploymentMetadata meta) {
        synchronized (mux) {
            List<SharedDeployment> deps = cache.get(meta.userVersion());

            if (deps != null) {
                assert !deps.isEmpty();

                for (SharedDeployment d : deps) {
                    if (d.hasParticipant(meta.senderNodeId(), meta.classLoaderId()))
                        return d;
                }
            }
        }

        return null;
    }

    /** {@inheritDoc} */
    @Override @Nullable public GridDeployment getDeployment(GridDeploymentMetadata meta) {
        assert meta != null;

        assert ctx.config().isPeerClassLoadingEnabled();

        // Validate metadata.
        assert meta.classLoaderId() != null;
        assert meta.senderNodeId() != null;
        assert meta.sequenceNumber() >= -1;
        assert meta.parentLoader() == null;

        if (log.isDebugEnabled())
            log.debug("Starting to peer-load class based on deployment metadata: " + meta);

        if (meta.participants() == null && !checkLoadRemoteClass(meta.className(), meta)) {
            if (log.isDebugEnabled())
                log.debug("Skipping deployment check as remote node does not have required class: " + meta);

            return null;
        }

        while (true) {
            List<SharedDeployment> depsToCheck = null;

            SharedDeployment dep = null;

            synchronized (mux) {
                // Check obsolete request.
                if (isDeadClassLoader(meta))
                    return null;

                if (!F.isEmpty(meta.participants())) {
                    Map<UUID, IgniteUuid> participants = new LinkedHashMap<>();

                    for (Map.Entry<UUID, IgniteUuid> e : meta.participants().entrySet()) {
                        // Warn if local node is in participants.
                        if (ctx.localNodeId().equals(e.getKey())) {
                            // Warn only if mode is not CONTINUOUS.
                            if (meta.deploymentMode() != CONTINUOUS)
                                LT.warn(log, "Local node is in participants (most probably, " +
                                    "IgniteConfiguration.getPeerClassLoadingLocalClassPathExclude() " +
                                    "is not used properly " +
                                    "[locNodeId=" + ctx.localNodeId() + ", meta=" + meta + ']');

                            continue;
                        }

                        // Skip left nodes.
                        if (ctx.discovery().node(e.getKey()) != null)
                            participants.put(e.getKey(), e.getValue());
                    }

                    if (!participants.isEmpty()) {
                        // Set new participants.
                        meta = new GridDeploymentMetadata(meta);

                        meta.participants(participants);

                        if (log.isDebugEnabled())
                            log.debug("Created new meta with updated participants: " + meta);
                    }
                    else {
                        if (log.isDebugEnabled())
                            log.debug("All participants has gone: " + meta);

                        // Should try locally cached deployments in CONTINUOUS mode because local node
                        // can have this classes deployed.
                        if (meta.deploymentMode() != CONTINUOUS)
                            return null;
                    }
                }
                else if (ctx.discovery().node(meta.senderNodeId()) == null) {
                    if (log.isDebugEnabled())
                        log.debug("Sender node has gone: " + meta);

                    return null;
                }

                dep = (SharedDeployment)searchDeploymentCache(meta);

                if (dep == null) {
                    List<SharedDeployment> deps = cache.get(meta.userVersion());

                    if (deps != null) {
                        assert !deps.isEmpty();

                        checkRedeploy(meta);

                        // Find existing deployments that need to be checked
                        // whether they should be reused for this request.
                        if (ctx.config().getDeploymentMode() == CONTINUOUS) {
                            for (SharedDeployment d : deps) {
                                if (!d.pendingUndeploy() && !d.undeployed()) {
                                    Map<UUID, IgniteUuid> parties = d.participants();

                                    if (parties != null) {
                                        IgniteUuid ldrId = parties.get(meta.senderNodeId());

                                        if (ldrId != null) {
                                            assert !ldrId.equals(meta.classLoaderId());

                                            if (log.isDebugEnabled())
                                                log.debug("Skipping deployment (loaders on remote node are different) " +
                                                    "[dep=" + d + ", meta=" + meta + ']');

                                            continue;
                                        }
                                    }

                                    if (depsToCheck == null)
                                        depsToCheck = new LinkedList<>();

                                    if (log.isDebugEnabled())
                                        log.debug("Adding deployment to check: " + d);

                                    depsToCheck.add(d);
                                }
                            }
                        }

                        // If no deployment can be reused, create a new one.
                        if (depsToCheck == null) {
                            dep = createNewDeployment(meta, false);

                            deps.add(dep);
                        }
                    }
                    else {
                        checkRedeploy(meta);

                        // Create peer class loader.
                        dep = createNewDeployment(meta, true);
                    }
                }
            }

            if (dep != null) {
                if (log.isDebugEnabled())
                    log.debug("Found SHARED or CONTINUOUS deployment after first check: " + dep);

                // Cache the deployed class.
                IgniteBiTuple<Class<?>, Throwable> cls = dep.deployedClass(meta.className(), meta.alias());

                if (cls.get1() == null) {
                    U.warn(log, "Failed to load peer class (ignore if class got undeployed during preloading) [alias=" +
                        meta.alias() + ", dep=" + dep + ']', cls.get2());

                    return null;
                }

                return dep;
            }

            assert meta.parentLoader() == null;
            assert depsToCheck != null;
            assert !depsToCheck.isEmpty();

            /*
             * Logic below must be performed outside of synchronization
             * because it involves network calls.
             */

            // Check if class can be loaded from existing nodes.
            // In most cases this loop will find something.
            for (SharedDeployment d : depsToCheck) {
                // Load class. Note, that remote node will not load this class.
                // The class will only be loaded on this node.
                Class<?> cls = d.deployedClass(meta.className(), meta.alias()).get1();

                if (cls != null) {
                    synchronized (mux) {
                        if (!d.undeployed() && !d.pendingUndeploy()) {
                            if (!addParticipant(d, meta))
                                return null;

                            if (log.isDebugEnabled())
                                log.debug("Acquired deployment after verifying it's availability on " +
                                    "existing nodes [depCls=" + cls + ", dep=" + d + ", meta=" + meta + ']');

                            return d;
                        }
                    }
                }
                else if (log.isDebugEnabled()) {
                    log.debug("Deployment cannot be reused (class does not exist on participating nodes) [dep=" + d +
                        ", meta=" + meta + ']');
                }
            }

            // We are here either because all participant nodes failed
            // or the class indeed should have a separate deployment.
            for (SharedDeployment d : depsToCheck) {
                // We check if any random class from existing deployment can be
                // loaded from sender node. If it can, then we reuse existing
                // deployment.
                if (meta.participants() != null || checkLoadRemoteClass(d.sampleClassName(), meta)) {
                    synchronized (mux) {
                        if (d.undeployed() || d.pendingUndeploy())
                            continue;

                        // Add new node prior to loading the class, so we attempt
                        // to load the class from the latest node.
                        if (!addParticipant(d, meta)) {
                            if (log.isDebugEnabled())
                                log.debug("Failed to add participant to deployment " +
                                    "[meta=" + meta + ", dep=" + dep + ']');

                            return null;
                        }
                    }

                    IgniteBiTuple<Class<?>, Throwable> depCls = d.deployedClass(meta.className(), meta.alias());

                    if (depCls.get1() == null) {
                        U.error(log, "Failed to peer load class after loading it as a resource [alias=" +
                            meta.alias() + ", dep=" + dep + ']', depCls.get2());

                        return null;
                    }

                    if (log.isDebugEnabled())
                        log.debug("Acquired deployment class after verifying other class " +
                            "availability on sender node [depCls=" + depCls + ", rndCls=" + d.sampleClassName() +
                            ", sampleClsName=" + d.sampleClassName() + ", meta=" + meta + ']');

                    return d;
                }
                else if (log.isDebugEnabled())
                    log.debug("Deployment cannot be reused (random class could not be loaded from sender node) [dep=" +
                        d + ", meta=" + meta + ']');
            }

            synchronized (mux) {
                if (log.isDebugEnabled())
                    log.debug("None of the existing class-loaders fits (will try to create a new one): " + meta);

                // Check obsolete request.
                if (isDeadClassLoader(meta))
                    return null;

                // Check that deployment picture has not changed.
                List<SharedDeployment> deps = cache.get(meta.userVersion());

                if (deps != null) {
                    assert !deps.isEmpty();

                    boolean retry = false;

                    for (SharedDeployment d : deps) {
                        // Double check if sender was already added.
                        if (d.hasParticipant(meta.senderNodeId(), meta.classLoaderId())) {
                            dep = d;

                            retry = false;

                            break;
                        }

                        // New deployment was added while outside of synchronization.
                        // Need to recheck it again.
                        if (!d.pendingUndeploy() && !d.undeployed() && !depsToCheck.contains(d)) {
                            Map<UUID, IgniteUuid> parties = d.participants();

                            if (parties == null || parties.get(meta.senderNodeId()) == null)
                                retry = true;
                        }
                    }

                    if (retry) {
                        if (log.isDebugEnabled())
                            log.debug("Retrying due to concurrency issues: " + meta);

                        // Outer while loop.
                        continue;
                    }

                    if (dep == null) {
                        // No new deployments were added, so we can safely add ours.
                        dep = createNewDeployment(meta, false);

                        deps.add(dep);

                        if (log.isDebugEnabled())
                            log.debug("Adding new deployment within second check [dep=" + dep + ", meta=" + meta + ']');
                    }
                }
                else {
                    dep = createNewDeployment(meta, true);

                    if (log.isDebugEnabled())
                        log.debug("Created new deployment within second check [dep=" + dep + ", meta=" + meta + ']');
                }
            }

            if (dep != null) {
                // Cache the deployed class.
                IgniteBiTuple<Class<?>, Throwable> cls = dep.deployedClass(meta.className(), meta.alias());

                if (cls.get1() == null) {
                    U.warn(log, "Failed to load peer class (ignore if class got undeployed during preloading) [alias=" +
                        meta.alias() + ", dep=" + dep + ']', cls.get2());

                    return null;
                }
            }

            return dep;
        }
    }

    /** {@inheritDoc} */
    @Override public void addParticipants(Map<UUID, IgniteUuid> allParticipants,
        Map<UUID, IgniteUuid> addedParticipants) {
        synchronized (mux) {
            for (List<SharedDeployment> deps : cache.values()) {
                for (SharedDeployment dep : deps) {
                    if (dep.undeployed() || dep.pendingUndeploy() || dep.deployMode() == CONTINUOUS)
                        continue;

                    if (hasAnyParticipant(dep, allParticipants)) {
                        for (Map.Entry<UUID, IgniteUuid> added : addedParticipants.entrySet()) {
                            UUID nodeId = added.getKey();

                            if (ctx.discovery().node(nodeId) == null)
                                continue; // For.

                            if (!dep.hasParticipant(nodeId, added.getValue()) &&
                                dep.addParticipant(nodeId, added.getValue())) {
                                if (log.isDebugEnabled())
                                    log.debug("Explicitly added participant [dep=" + dep + ", nodeId=" + nodeId +
                                        ", ldrId=" + added.getValue() + ']');
                            }
                        }
                    }
                }
            }
        }
    }

    /**
     * Checks if given deployment contains any of given participants.
     *
     * @param dep Deployment to check.
     * @param participants Participants to check.
     * @return {@code True} if deployment contains any of given participants.
     */
    private boolean hasAnyParticipant(SharedDeployment dep, Map<UUID, IgniteUuid> participants) {
        assert Thread.holdsLock(mux);

        for (Map.Entry<UUID, IgniteUuid> entry : participants.entrySet()) {
            if (dep.hasParticipant(entry.getKey(), entry.getValue()))
                return true;
        }

        return false;
    }

    /**
     * Checks that given class can be loaded from remote node specified by sender ID in metadata.
     *
     * @param clsName Class name to try load.
     * @param meta Deployment metadata.
     * @return {@code True} if class do exist on remote node, {@code false} otherwise.
     */
    private boolean checkLoadRemoteClass(String clsName, GridDeploymentMetadata meta) {
        assert clsName != null;
        assert meta != null;
        assert meta.participants() == null;

        // First check resource cache.
        Map<String, Boolean> ldrRsrcCache = rsrcCache.get(meta.classLoaderId());

        if (ldrRsrcCache != null) {
            Boolean res = ldrRsrcCache.get(clsName);

            if (res != null)
                return res;
        }

        // Check dead class loaders.
        if (deadClsLdrs.contains(meta.classLoaderId()))
            return false;

        int lockId = clsName.hashCode() * 31 + meta.classLoaderId().hashCode();

        loadRmtLock.lock(lockId);

        try {
            // Check once more inside lock.
            ldrRsrcCache = rsrcCache.get(meta.classLoaderId());

            if (ldrRsrcCache != null) {
                Boolean res = ldrRsrcCache.get(clsName);

                if (res != null)
                    return res;
            }

            // Check dead class loaders.
            if (deadClsLdrs.contains(meta.classLoaderId()))
                return false;

            // Temporary class loader.
            GridDeploymentClassLoader temp = new GridDeploymentClassLoader(
                IgniteUuid.fromUuid(ctx.localNodeId()),
                meta.userVersion(),
                meta.deploymentMode(),
                true,
                ctx,
                ctx.config().getClassLoader() != null ? ctx.config().getClassLoader() : U.gridClassLoader(),
                meta.classLoaderId(),
                meta.senderNodeId(),
                comm,
                ctx.config().getNetworkTimeout(),
                log,
                ctx.config().getPeerClassLoadingLocalClassPathExclude(),
                0,
                false,
                true);

            String path = U.classNameToResourceName(clsName);

            // Check if specified class can be loaded from sender node.
            InputStream rsrcIn = null;

            try {
                boolean timeout = false;

                try {
                    rsrcIn = temp.getResourceAsStreamEx(path);
                }
                catch (TimeoutException e) {
                    timeout = true;
                }

                boolean found = rsrcIn != null;

                // Add result to resource cache.
                // If missed resource cache is disabled and resource not found
                // we do not save the result (only hits are saved in this case).
                if (found || missedRsrcCacheSize > 0) {
                    if (ldrRsrcCache == null)
                        ldrRsrcCache = F.addIfAbsent(rsrcCache, meta.classLoaderId(),
                            new ConcurrentHashMap<String, Boolean>());

                    // This is the only place where cache could have been changed,
                    // so we remove only here if classloader have been undeployed
                    // concurrently.
                    if (deadClsLdrs.contains(meta.classLoaderId())) {
                        rsrcCache.remove(meta.classLoaderId());

                        return false;
                    }
                    else if (!timeout)
                        // Cache result if classloader is still alive.
                        ldrRsrcCache.put(clsName, found);
                }

                return found;
            }
            finally {
                // We don't need the actual stream.
                U.closeQuiet(rsrcIn);
            }
        }
        finally {
            loadRmtLock.unlock(lockId);
        }
    }

    /**
     * Records all undeployed tasks.
     *
     * @param nodeId Left node ID.
     * @param undeployed Undeployed deployments.
     */
    private void recordUndeployed(@Nullable UUID nodeId, Collection<SharedDeployment> undeployed) {
        if (!F.isEmpty(undeployed))
            for (SharedDeployment d : undeployed)
                d.recordUndeployed(nodeId);
    }

    /**
     * @param meta Request metadata.
     * @return {@code True} if class loader is obsolete.
     */
    private boolean isDeadClassLoader(GridDeploymentMetadata meta) {
        assert Thread.holdsLock(mux);

        if (deadClsLdrs.contains(meta.classLoaderId())) {
            if (log.isDebugEnabled())
                log.debug("Ignoring request for obsolete class loader: " + meta);

            return true;
        }

        return false;
    }

    /**
     * Adds new participant to deployment.
     *
     * @param dep Shared deployment.
     * @param meta Request metadata.
     * @return {@code True} if participant was added.
     */
    private boolean addParticipant(SharedDeployment dep, GridDeploymentMetadata meta) {
        assert dep != null;
        assert meta != null;

        assert Thread.holdsLock(mux);

        if (!checkModeMatch(dep, meta))
            return false;

        if (meta.participants() != null) {
            for (Map.Entry<UUID, IgniteUuid> e : meta.participants().entrySet()) {
                if (ctx.discovery().node(e.getKey()) != null) {
                    dep.addParticipant(e.getKey(), e.getValue());

                    if (log.isDebugEnabled())
                        log.debug("Added new participant [nodeId=" + e.getKey() + ", clsLdrId=" + e.getValue() +
                            ", seqNum=" + e.getValue().localId() + ']');
                }
                else if (log.isDebugEnabled())
                    log.debug("Skipped participant (node left?) [nodeId=" + e.getKey() +
                        ", clsLdrId=" + e.getValue() + ", seqNum=" + e.getValue().localId() + ']');
            }
        }

        if (dep.deployMode() == CONTINUOUS || meta.participants() == null) {
            if (!dep.addParticipant(meta.senderNodeId(), meta.classLoaderId())) {
                U.warn(log, "Failed to create shared mode deployment " +
                    "(requested class loader was already undeployed, did sender node leave grid?) " +
                    "[clsLdrId=" + meta.classLoaderId() + ", senderNodeId=" + meta.senderNodeId() + ']');

                return false;
            }

            if (log.isDebugEnabled())
                log.debug("Added new participant [nodeId=" + meta.senderNodeId() + ", clsLdrId=" +
                    meta.classLoaderId() + ", seqNum=" + meta.sequenceNumber() + ']');
        }

        return true;
    }

    /**
     * Checks if deployment modes match.
     *
     * @param dep Shared deployment.
     * @param meta Request metadata.
     * @return {@code True} if shared deployment modes match.
     */
    private boolean checkModeMatch(GridDeploymentInfo dep, GridDeploymentMetadata meta) {
        if (dep.deployMode() != meta.deploymentMode()) {
            U.warn(log, "Received invalid deployment mode (will not deploy, make sure that all nodes " +
                "executing the same classes in shared mode have identical GridDeploymentMode parameter) [mode=" +
                meta.deploymentMode() + ", expected=" + dep.deployMode() + ']');

            return false;
        }

        return true;
    }

    /**
     * Removes obsolete deployments in case of redeploy.
     *
     * @param meta Request metadata.
     */
    private void checkRedeploy(GridDeploymentMetadata meta) {
        assert Thread.holdsLock(mux);

        for (List<SharedDeployment> deps : cache.values()) {
            for (SharedDeployment dep : deps) {
                if (!dep.undeployed() && !dep.pendingUndeploy()) {
                    long undeployTimeout = ctx.config().getNetworkTimeout();

                    // Only check deployments with no participants.
                    if (!dep.hasParticipants() &&
                        dep.deployMode() == CONTINUOUS &&
                        dep.existingDeployedClass(meta.className()) != null &&
                        !meta.userVersion().equals(dep.userVersion())) {
                        // In case of SHARED deployment it is possible to get hear if
                        // unmarshalling happens during undeploy. In this case, we
                        // simply don't do anything.
                        // Change from shared deploy to shared undeploy or user version change.
                        // Simply remove all deployments with no participating nodes.
                        dep.onUndeployScheduled();

                        if (log.isDebugEnabled())
                            log.debug("Deployment was scheduled for undeploy: " + dep);

                        // Lifespan time.
                        final long endTime = U.currentTimeMillis() + undeployTimeout;

                        // Deployment to undeploy.
                        final SharedDeployment undep = dep;

                        if (endTime > 0) {
                            ctx.timeout().addTimeoutObject(new GridTimeoutObject() {
                                @Override public IgniteUuid timeoutId() {
                                    return undep.classLoaderId();
                                }

                                @Override public long endTime() {
                                    return endTime < 0 ? Long.MAX_VALUE : endTime;
                                }

                                @Override public void onTimeout() {
                                    boolean rmv = false;

                                    // Hot redeployment.
                                    synchronized (mux) {
                                        assert undep.pendingUndeploy();

                                        if (!undep.undeployed()) {
                                            undep.undeploy();

                                            undep.onRemoved();

                                            rmv = true;

                                            Collection<SharedDeployment> deps = cache.get(undep.userVersion());

                                            if (deps != null) {
                                                for (Iterator<SharedDeployment> i = deps.iterator(); i.hasNext();)
                                                    if (i.next() == undep)
                                                        i.remove();

                                                if (deps.isEmpty())
                                                    cache.remove(undep.userVersion());
                                            }

                                            if (log.isInfoEnabled())
                                                log.info("Undeployed class loader due to deployment mode change, " +
                                                    "user version change, or hot redeployment: " + undep);
                                        }
                                    }

                                    // Outside synchronization.
                                    if (rmv)
                                        undep.recordUndeployed(null);
                                }
                            });
                        }
                    }
                }
            }
        }
    }

    /** {@inheritDoc} */
    @Override public void explicitUndeploy(UUID nodeId, String rsrcName) {
        Collection<SharedDeployment> undeployed = new LinkedList<>();

        synchronized (mux) {
            for (Iterator<List<SharedDeployment>> i1 = cache.values().iterator(); i1.hasNext();) {
                List<SharedDeployment> deps = i1.next();

                for (Iterator<SharedDeployment> i2 = deps.iterator(); i2.hasNext();) {
                    SharedDeployment dep = i2.next();

                    if (dep.hasName(rsrcName)) {
                        if (!dep.undeployed()) {
                            dep.undeploy();

                            dep.onRemoved();

                            // Undeploy.
                            i2.remove();

                            undeployed.add(dep);

                            if (log.isInfoEnabled())
                                log.info("Undeployed per-version class loader: " + dep);
                        }

                        break;
                    }
                }

                if (deps.isEmpty())
                    i1.remove();
            }
        }

        recordUndeployed(null, undeployed);
    }

    /**
     * Creates and caches new deployment.
     *
     * @param meta Deployment metadata.
     * @param isCache Whether or not to cache.
     * @return New deployment.
     */
    private SharedDeployment createNewDeployment(GridDeploymentMetadata meta, boolean isCache) {
        assert Thread.holdsLock(mux);

        assert meta.parentLoader() == null;

        IgniteUuid ldrId = IgniteUuid.fromUuid(ctx.localNodeId());

        GridDeploymentClassLoader clsLdr;

        if (meta.deploymentMode() == CONTINUOUS || meta.participants() == null) {
            // Create peer class loader.
            // Note that we are passing empty list for local P2P exclude, as it really
            // does not make sense with shared deployment.
            clsLdr = new GridDeploymentClassLoader(
                ldrId,
                meta.userVersion(),
                meta.deploymentMode(),
                false,
                ctx,
                ctx.config().getClassLoader() != null ? ctx.config().getClassLoader() : U.gridClassLoader(),
                meta.classLoaderId(),
                meta.senderNodeId(),
                comm,
                ctx.config().getNetworkTimeout(),
                log,
                ctx.config().getPeerClassLoadingLocalClassPathExclude(),
                ctx.config().getPeerClassLoadingMissedResourcesCacheSize(),
                meta.deploymentMode() == CONTINUOUS /* enable class byte cache in CONTINUOUS mode */,
                false);

            if (meta.participants() != null)
                for (Map.Entry<UUID, IgniteUuid> e : meta.participants().entrySet())
                    clsLdr.register(e.getKey(), e.getValue());

            if (log.isDebugEnabled())
                log.debug("Created class loader in CONTINUOUS mode or without participants " +
                    "[ldr=" + clsLdr + ", meta=" + meta + ']');
        }
        else {
            assert meta.deploymentMode() == SHARED;

            // Create peer class loader.
            // Note that we are passing empty list for local P2P exclude, as it really
            // does not make sense with shared deployment.
            clsLdr = new GridDeploymentClassLoader(
                ldrId,
                meta.userVersion(),
                meta.deploymentMode(),
                false,
                ctx,
                U.gridClassLoader(),
                meta.participants(),
                comm,
                ctx.config().getNetworkTimeout(),
                log,
                ctx.config().getPeerClassLoadingLocalClassPathExclude(),
                ctx.config().getPeerClassLoadingMissedResourcesCacheSize(),
                false,
                false);

            if (log.isDebugEnabled())
                log.debug("Created classloader in SHARED mode with participants " +
                    "[ldr=" + clsLdr + ", meta=" + meta + ']');
        }

        // Give this deployment a unique class loader to emphasize that this
        // ID is unique to this shared deployment and is not ID of loader on
        // sender node.
        SharedDeployment dep = new SharedDeployment(meta.deploymentMode(), clsLdr, ldrId,
            meta.userVersion(), meta.alias());

        if (log.isDebugEnabled())
            log.debug("Created new deployment: " + dep);

        if (isCache) {
            List<SharedDeployment> deps = F.addIfAbsent(cache, meta.userVersion(), new LinkedList<SharedDeployment>());

            assert deps != null;

            deps.add(dep);

            if (log.isDebugEnabled())
                log.debug("Added deployment to cache: " + cache);
        }

        return dep;
    }

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

    /**
     *
     */
    private class SharedDeployment extends GridDeployment {
        /** Flag indicating whether this deployment was removed from cache. */
        private boolean rmv;

        /**
         * @param depMode Deployment mode.
         * @param clsLdr Class loader.
         * @param clsLdrId Class loader ID.
         * @param userVer User version.
         * @param sampleClsName Sample class name.
         */
        SharedDeployment(DeploymentMode depMode,
            GridDeploymentClassLoader clsLdr, IgniteUuid clsLdrId,
            String userVer, String sampleClsName) {
            super(depMode, clsLdr, clsLdrId, userVer, sampleClsName, false);
        }

        /** {@inheritDoc} */
        @Override public GridDeploymentClassLoader classLoader() {
            return (GridDeploymentClassLoader)super.classLoader();
        }

        /**
         * @param nodeId Grid node ID.
         * @param ldrId Class loader ID.
         * @return Whether actually added or not.
         */
        boolean addParticipant(UUID nodeId, IgniteUuid ldrId) {
            assert nodeId != null;
            assert ldrId != null;

            assert Thread.holdsLock(mux);

            if (!deadClsLdrs.contains(ldrId)) {
                classLoader().register(nodeId, ldrId);

                return true;
            }

            return false;
        }

        /**
         * @param nodeId Node ID to remove.
         */
        void removeParticipant(UUID nodeId) {
            assert nodeId != null;

            assert Thread.holdsLock(mux);

            IgniteUuid ldrId = classLoader().unregister(nodeId);

            if (log.isDebugEnabled())
                log.debug("Registering dead class loader ID: " + ldrId);

            if (ldrId == null)
                return;

            deadClsLdrs.add(ldrId);

            // Remove hits/misses from resource cache as well.
            rsrcCache.remove(ldrId);
        }

        /**
         * @return Set of participating nodes.
         */
        Collection<UUID> getParticipantNodeIds() {
            assert Thread.holdsLock(mux);

            return classLoader().registeredNodeIds();
        }

        /**
         * @param nodeId Node ID.
         * @return Class loader ID for node ID.
         */
        IgniteUuid getClassLoaderId(UUID nodeId) {
            assert nodeId != null;

            assert Thread.holdsLock(mux);

            return classLoader().registeredClassLoaderId(nodeId);
        }

        /**
         * @return Registered class loader IDs.
         */
        Collection<IgniteUuid> getClassLoaderIds() {
            assert Thread.holdsLock(mux);

            return classLoader().registeredClassLoaderIds();
        }


        /**
         * @return {@code True} if deployment has any node participants.
         */
        boolean hasParticipants() {
            assert Thread.holdsLock(mux);

            return classLoader().hasRegisteredNodes();
        }

        /**
         * Checks if node is participating in deployment.
         *
         * @param nodeId Node ID to check.
         * @param ldrId Class loader ID.
         * @return {@code True} if node is participating in deployment.
         */
        boolean hasParticipant(UUID nodeId, IgniteUuid ldrId) {
            assert nodeId != null;
            assert ldrId != null;

            return classLoader().hasRegisteredNode(nodeId, ldrId);
        }

        /**
         * Gets property removed.
         *
         * @return Property removed.
         */
        boolean isRemoved() {
            assert Thread.holdsLock(mux);

            return rmv;
        }

        /**
         * Sets property removed.
         */
        void onRemoved() {
            assert Thread.holdsLock(mux);

            rmv = true;

            Collection<IgniteUuid> deadIds = classLoader().registeredClassLoaderIds();

            if (log.isDebugEnabled())
                log.debug("Registering dead class loader IDs: " + deadIds);

            deadClsLdrs.addAll(deadIds);

            // Remove hits/misses from resource cache as well.
            for (IgniteUuid clsLdrId : deadIds)
                rsrcCache.remove(clsLdrId);
        }

        /** {@inheritDoc} */
        @Override public void onDeployed(Class<?> cls) {
            assert !Thread.holdsLock(mux);

            boolean isTask = isTask(cls);

            String msg = (isTask ? "Task" : "Class") + " was deployed in SHARED or CONTINUOUS mode: " + cls;

            int type = isTask ? EVT_TASK_DEPLOYED : EVT_CLASS_DEPLOYED;

            if (ctx.event().isRecordable(type)) {
                DeploymentEvent evt = new DeploymentEvent();

                evt.node(ctx.discovery().localNode());
                evt.message(msg);
                evt.type(type);
                evt.alias(cls.getName());

                ctx.event().record(evt);
            }

            if (log.isInfoEnabled())
                log.info(msg);
        }

        /**
         * Called to record all undeployed classes..
         *
         * @param leftNodeId Left node ID.
         */
        void recordUndeployed(@Nullable UUID leftNodeId) {
            assert !Thread.holdsLock(mux);

            for (Map.Entry<String, Class<?>> depCls : deployedClassMap().entrySet()) {
                boolean isTask = isTask(depCls.getValue());

                String msg = (isTask ? "Task" : "Class") + " was undeployed in SHARED or CONTINUOUS mode " +
                    "[cls=" + depCls.getValue() + ", alias=" + depCls.getKey() + ']';

                int type = isTask ? EVT_TASK_UNDEPLOYED : EVT_CLASS_UNDEPLOYED;

                if (ctx.event().isRecordable(type)) {
                    DeploymentEvent evt = new DeploymentEvent();

                    evt.node(ctx.discovery().localNode());
                    evt.message(msg);
                    evt.type(type);
                    evt.alias(depCls.getKey());

                    ctx.event().record(evt);
                }

                if (log.isInfoEnabled())
                    log.info(msg);
            }

            if (obsolete()) {
                // Resource cleanup.
                ctx.resource().onUndeployed(this);

                ClassLoader ldr = classLoader();

                ctx.cache().onUndeployed(ldr);

               // Clear static class cache.
                U.clearClassFromClassCache(ctx.cache().context().deploy().globalLoader(), sampleClassName());

                for (String alias : deployedClassMap().keySet())
                    U.clearClassFromClassCache(ctx.cache().context().deploy().globalLoader(), alias);

                // Clear optimized marshaller's cache.
                if (ctx.config().getMarshaller() instanceof AbstractMarshaller)
                    ((AbstractMarshaller)ctx.config().getMarshaller()).onUndeploy(ldr);

                clearSerializationCaches();

                // Class loader cache should be cleared in the last order.
                GridAnnotationsCache.onUndeployed(ldr);
                GridClassLoaderCache.onUndeployed(ldr);
            }
        }

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