modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentManager.java - ignite - Git at Google

 /*
  * 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.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.events.DiscoveryEvent;
 import org.apache.ignite.failure.FailureContext;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
 import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
 import org.apache.ignite.internal.events.DiscoveryCustomEvent;
 import org.apache.ignite.internal.managers.communication.GridMessageListener;
 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.managers.eventstorage.HighPriorityListener;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
 import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.internal.util.worker.GridWorker;
 import org.apache.ignite.thread.IgniteThread;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;

 import static org.apache.ignite.IgniteSystemProperties.IGNITE_LONG_OPERATIONS_DUMP_TIMEOUT_LIMIT;
 import static org.apache.ignite.IgniteSystemProperties.getLong;
 import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
 import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
 import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
 import static org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
 import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
 import static org.apache.ignite.internal.GridTopic.TOPIC_SERVICES;
 import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT;

 /**
  * Service deployment manager.
  *
  * @see ServiceDeploymentTask
  * @see ServiceDeploymentActions
  */
 public class ServiceDeploymentManager {
     /** Busy lock. */
     private final GridSpinBusyLock busyLock = new GridSpinBusyLock();

     /** Services discovery messages listener. */
     private final DiscoveryEventListener discoLsnr = new ServiceDiscoveryListener();

     /** Services communication messages listener. */
     private final GridMessageListener commLsnr = new ServiceCommunicationListener();

     /** Services deployments tasks. */
     private final Map<ServiceDeploymentProcessId, ServiceDeploymentTask> tasks = new ConcurrentHashMap<>();

     /** Discovery events received while cluster state transition was in progress. */
     private final List<PendingEventHolder> pendingEvts = new ArrayList<>();

     /** Topology version of latest deployment task's event. */
     private final AtomicReference<AffinityTopologyVersion> readyTopVer =
         new AtomicReference<>(AffinityTopologyVersion.NONE);

     /** Kernal context. */
     private final GridKernalContext ctx;

     /** Logger. */
     private final IgniteLogger log;

     /** Deployment worker. */
     private final ServicesDeploymentWorker depWorker;

     /** Default dump operation limit. */
     private final long dfltDumpTimeoutLimit;

     /**
      * @param ctx Grid kernal context.
      */
     ServiceDeploymentManager(@NotNull GridKernalContext ctx) {
         this.ctx = ctx;

         log = ctx.log(getClass());

         ctx.event().addDiscoveryEventListener(discoLsnr,
             EVT_NODE_JOINED, EVT_NODE_LEFT, EVT_NODE_FAILED, EVT_DISCOVERY_CUSTOM_EVT);

         ctx.io().addMessageListener(TOPIC_SERVICES, commLsnr);

         depWorker = new ServicesDeploymentWorker();

         long limit = getLong(IGNITE_LONG_OPERATIONS_DUMP_TIMEOUT_LIMIT, 0);

         dfltDumpTimeoutLimit = limit <= 0 ? 30 * 60_000 : limit;
     }

     /**
      * Starts processing of services deployments tasks.
      */
     void startProcessing() {
         assert depWorker.runner() == null : "Method shouldn't be called twice during lifecycle;";

         new IgniteThread(ctx.igniteInstanceName(), "services-deployment-worker", depWorker).start();
     }

     /**
      * Stops processing of services deployments tasks.
      *
      * @param stopErr Cause error of deployment manager stop.
      */
     void stopProcessing(IgniteCheckedException stopErr) {
         busyLock.block();

         try {
             ctx.event().removeDiscoveryEventListener(discoLsnr);

             ctx.io().removeMessageListener(commLsnr);

             U.cancel(depWorker);

             U.join(depWorker, log);

             depWorker.tasksQueue.clear();

             pendingEvts.clear();

             tasks.values().forEach(t -> t.completeError(stopErr));

             tasks.clear();
         }
         finally {
             busyLock.unblock();
         }
     }

     /**
      * @return Ready topology version.
      */
     public AffinityTopologyVersion readyTopologyVersion() {
         return readyTopVer.get();
     }

     /**
      * Special handler for local discovery events for which the regular events are not generated, e.g. local join and
      * client reconnect events.
      *
      * @param evt Discovery event.
      * @param discoCache Discovery cache.
      * @param depActions Service deployment actions.
      */
     void onLocalJoin(DiscoveryEvent evt, DiscoCache discoCache, ServiceDeploymentActions depActions) {
         checkClusterStateAndAddTask(evt, discoCache, depActions);
     }

     /**
      * Invokes {@link GridWorker#blockingSectionBegin()} for service deployment worker.
      * <p/>
      * Should be called from service deployment worker thread.
      */
     void deployerBlockingSectionBegin() {
         assert depWorker != null && Thread.currentThread() == depWorker.runner();

         depWorker.blockingSectionBegin();
     }

     /**
      * Invokes {@link GridWorker#blockingSectionEnd()} for service deployment worker.
      * <p/>
      * Should be called from service deployment worker thread.
      */
     void deployerBlockingSectionEnd() {
         assert depWorker != null && Thread.currentThread() == depWorker.runner();

         depWorker.blockingSectionEnd();
     }

     /**
      * Checks cluster state and handles given event.
      * <pre>
      * - if cluster is active, then adds event in deployment queue;
      * - if cluster state in transition, them adds to pending events;
      * - if cluster is inactive, then ignore event;
      * </pre>
      * <b>Should be called from discovery thread.</b>
      *
      * @param evt Discovery event.
      * @param discoCache Discovery cache.
      * @param depActions Services deployment actions.
      */
     private void checkClusterStateAndAddTask(@NotNull DiscoveryEvent evt, @NotNull DiscoCache discoCache,
         @Nullable ServiceDeploymentActions depActions) {
         if (discoCache.state().transition())
             pendingEvts.add(new PendingEventHolder(evt, discoCache.version(), depActions));
         else if (discoCache.state().active())
             addTask(evt, discoCache.version(), depActions);
         else if (log.isDebugEnabled())
             log.debug("Ignore event, cluster is inactive, evt=" + evt);
     }

     /**
      * Adds deployment task with given deployment process id.
      * </p>
      * <b>Should be called from discovery thread.</b>
      *
      * @param evt Discovery event.
      * @param topVer Topology version.
      * @param depActions Services deployment actions.
      */
     private void addTask(@NotNull DiscoveryEvent evt, @NotNull AffinityTopologyVersion topVer,
         @Nullable ServiceDeploymentActions depActions) {
         final ServiceDeploymentProcessId depId = deploymentId(evt, topVer);

         ServiceDeploymentTask task = tasks.computeIfAbsent(depId,
             t -> new ServiceDeploymentTask(ctx, depId));

         if (!task.onEnqueued()) {
             if (log.isDebugEnabled()) {
                 log.debug("Service deployment process hasn't been started for discovery event, because of " +
                     "a task with the same deployment process id is already added (possible cause is message's" +
                     " double delivering), evt=" + evt);
             }

             return;
         }

         assert task.event() == null && task.topologyVersion() == null;

         task.onEvent(evt, topVer, depActions);

         depWorker.tasksQueue.add(task);
     }

     /**
      * Creates service deployment process id.
      *
      * @param evt Discovery event.
      * @param topVer Topology version.
      * @return Services deployment process id.
      */
     private ServiceDeploymentProcessId deploymentId(@NotNull DiscoveryEvent evt,
         @NotNull AffinityTopologyVersion topVer) {
         return evt instanceof DiscoveryCustomEvent ?
             new ServiceDeploymentProcessId(((DiscoveryCustomEvent)evt).customMessage().id()) :
             new ServiceDeploymentProcessId(topVer);
     }

     /**
      * Clones some instances of {@link DiscoveryCustomEvent} to capture necessary data, to avoid custom messages's
      * nullifying by {@link GridDhtPartitionsExchangeFuture#onDone}.
      *
      * @param evt Discovery event.
      * @return Discovery event to process.
      */
     private DiscoveryCustomEvent copyIfNeeded(@NotNull DiscoveryCustomEvent evt) {
         DiscoveryCustomMessage msg = evt.customMessage();

         assert msg != null : "DiscoveryCustomMessage has been nullified concurrently, evt=" + evt;

         if (msg instanceof ServiceChangeBatchRequest)
             return evt;

         DiscoveryCustomEvent cp = new DiscoveryCustomEvent();

         cp.node(evt.node());
         cp.customMessage(msg);
         cp.eventNode(evt.eventNode());
         cp.affinityTopologyVersion(evt.affinityTopologyVersion());

         return cp;
     }

     /**
      * Services discovery messages high priority listener.
      * <p/>
      * The listener should be notified earlier then PME's listener because of a custom message of {@link
      * DiscoveryCustomEvent} may be nullified in PME before the listener will be able to capture it.
      */
     private class ServiceDiscoveryListener implements DiscoveryEventListener, HighPriorityListener {
         /** {@inheritDoc} */
         @Override public void onEvent(final DiscoveryEvent evt, final DiscoCache discoCache) {
             if (!enterBusy())
                 return;

             try {
                 final UUID snd = evt.eventNode().id();
                 final int evtType = evt.type();

                 assert snd != null : "Event's node id shouldn't be null.";
                 assert evtType == EVT_NODE_JOINED || evtType == EVT_NODE_LEFT || evtType == EVT_NODE_FAILED
                     || evtType == EVT_DISCOVERY_CUSTOM_EVT : "Unexpected event was received, evt=" + evt;

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

                     if (msg instanceof ChangeGlobalStateFinishMessage) {
                         ChangeGlobalStateFinishMessage msg0 = (ChangeGlobalStateFinishMessage)msg;

                         if (msg0.clusterActive())
                             pendingEvts.forEach(t -> addTask(t.evt, t.topVer, t.depActions));
                         else if (log.isDebugEnabled())
                             pendingEvts.forEach(t -> log.debug("Ignore event, cluster is inactive: " + t.evt));

                         pendingEvts.clear();
                     }
                     else {
                         if (msg instanceof ServiceClusterDeploymentResultBatch) {
                             ServiceClusterDeploymentResultBatch msg0 = (ServiceClusterDeploymentResultBatch)msg;

                             if (log.isDebugEnabled()) {
                                 log.debug("Received services full deployments message : " +
                                     "[locId=" + ctx.localNodeId() + ", snd=" + snd + ", msg=" + msg0 + ']');
                             }

                             ServiceDeploymentProcessId depId = msg0.deploymentId();

                             assert depId != null;

                             ServiceDeploymentTask task = tasks.get(depId);

                             if (task != null) // May be null in case of double delivering
                                 task.onReceiveFullDeploymentsMessage(msg0);
                         }
                         else if (msg instanceof CacheAffinityChangeMessage)
                             addTask(copyIfNeeded((DiscoveryCustomEvent)evt), discoCache.version(), null);
                         else {
                             ServiceDeploymentActions depActions = null;

                             if (msg instanceof ChangeGlobalStateMessage)
                                 depActions = ((ChangeGlobalStateMessage)msg).servicesDeploymentActions();
                             else if (msg instanceof ServiceChangeBatchRequest) {
                                 depActions = ((ServiceChangeBatchRequest)msg)
                                     .servicesDeploymentActions();
                             }
                             else if (msg instanceof DynamicCacheChangeBatch)
                                 depActions = ((DynamicCacheChangeBatch)msg).servicesDeploymentActions();

                             if (depActions != null)
                                 addTask(copyIfNeeded((DiscoveryCustomEvent)evt), discoCache.version(), depActions);
                         }
                     }
                 }
                 else {
                     if (evtType == EVT_NODE_LEFT || evtType == EVT_NODE_FAILED)
                         tasks.values().forEach(t -> t.onNodeLeft(snd));

                     checkClusterStateAndAddTask(evt, discoCache, null);
                 }
             }
             finally {
                 leaveBusy();
             }
         }

         /** {@inheritDoc} */
         @Override public int order() {
             return 0;
         }
     }

     /**
      * Pending event's holder.
      */
     private static class PendingEventHolder {
         /** Discovery event. */
         private DiscoveryEvent evt;

         /** Topology version. */
         private AffinityTopologyVersion topVer;

         /** Services deployemnt actions. */
         private ServiceDeploymentActions depActions;

         /**
          * @param evt Discovery event.
          * @param topVer Topology version.
          * @param depActions Services deployment actions.
          */
         private PendingEventHolder(DiscoveryEvent evt,
             AffinityTopologyVersion topVer, ServiceDeploymentActions depActions) {
             this.evt = evt;
             this.topVer = topVer;
             this.depActions = depActions;
         }
     }

     /**
      * Services messages communication listener.
      */
     private class ServiceCommunicationListener implements GridMessageListener {
         /** {@inheritDoc} */
         @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
             if (!enterBusy())
                 return;

             try {
                 if (msg instanceof ServiceSingleNodeDeploymentResultBatch) {
                     ServiceSingleNodeDeploymentResultBatch msg0 = (ServiceSingleNodeDeploymentResultBatch)msg;

                     if (log.isDebugEnabled()) {
                         log.debug("Received services single deployments message : " +
                             "[locId=" + ctx.localNodeId() + ", snd=" + nodeId + ", msg=" + msg0 + ']');
                     }

                     tasks.computeIfAbsent(msg0.deploymentId(),
                         t -> new ServiceDeploymentTask(ctx, msg0.deploymentId()))
                         .onReceiveSingleDeploymentsMessage(nodeId, msg0);
                 }
             }
             finally {
                 leaveBusy();
             }
         }
     }

     /**
      * Services deployment worker.
      */
     private class ServicesDeploymentWorker extends GridWorker {
         /** Queue to process. */
         private final LinkedBlockingQueue<ServiceDeploymentTask> tasksQueue = new LinkedBlockingQueue<>();

         /** {@inheritDoc} */
         private ServicesDeploymentWorker() {
             super(ctx.igniteInstanceName(), "services-deployment-worker",
                 ServiceDeploymentManager.this.log, ctx.workersRegistry());
         }

         /** {@inheritDoc} */
         @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException {
             Throwable err = null;

             try {
                 ServiceDeploymentTask task;

                 while (!isCancelled()) {
                     onIdle();

                     blockingSectionBegin();

                     try {
                         task = tasksQueue.take();
                     }
                     finally {
                         blockingSectionEnd();
                     }

                     if (isCancelled())
                         Thread.currentThread().interrupt();

                     task.init();

                     final long dumpTimeout = 2 * ctx.config().getNetworkTimeout();

                     long dumpCnt = 0;
                     long nextDumpTime = 0;

                     while (true) {
                         try {
                             blockingSectionBegin();

                             try {
                                 task.waitForComplete(dumpTimeout);
                             }
                             finally {
                                 blockingSectionEnd();
                             }

                             taskPostProcessing(task);

                             break;
                         }
                         catch (IgniteFutureTimeoutCheckedException ignored) {
                             if (isCancelled())
                                 return;

                             if (nextDumpTime <= U.currentTimeMillis()) {
                                 log.warning("Failed to wait service deployment process or timeout had been" +
                                     " reached, timeout=" + dumpTimeout +
                                     (log.isDebugEnabled() ? ", task=" + task : ", taskDepId=" + task.deploymentId()));

                                 long nextTimeout = dumpTimeout * (2 + dumpCnt++);

                                 nextDumpTime = U.currentTimeMillis() + Math.min(nextTimeout, dfltDumpTimeoutLimit);
                             }
                         }
                         catch (ClusterTopologyServerNotFoundException e) {
                             U.error(log, e);

                             taskPostProcessing(task);

                             break;
                         }
                     }
                 }
             }
             catch (InterruptedException | IgniteInterruptedCheckedException e) {
                 Thread.currentThread().interrupt();

                 if (!isCancelled())
                     err = e;
             }
             catch (Throwable t) {
                 err = t;
             }
             finally {
                 if (err == null && !isCancelled())
                     err = new IllegalStateException("Worker " + name() + " is terminated unexpectedly.");

                 if (err instanceof OutOfMemoryError)
                     ctx.failure().process(new FailureContext(CRITICAL_ERROR, err));
                 else if (err != null)
                     ctx.failure().process(new FailureContext(SYSTEM_WORKER_TERMINATION, err));
             }
         }

         /**
          * Does additional actions after task's completion.
          */
         private void taskPostProcessing(ServiceDeploymentTask task) {
             AffinityTopologyVersion readyVer = readyTopVer.get();

             readyTopVer.compareAndSet(readyVer, task.topologyVersion());

             tasks.remove(task.deploymentId());
         }
     }

     /**
      * Enters busy state.
      *
      * @return {@code true} if entered to busy state.
      */
     private boolean enterBusy() {
         return busyLock.enterBusy();
     }

     /**
      * Leaves busy state.
      */
     private void leaveBusy() {
         busyLock.leaveBusy();
     }
 }
	/*
	* 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.util.ArrayList;
	import java.util.List;
	import java.util.Map;
	import java.util.UUID;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.LinkedBlockingQueue;
	import java.util.concurrent.atomic.AtomicReference;
	import org.apache.ignite.IgniteCheckedException;
	import org.apache.ignite.IgniteLogger;
	import org.apache.ignite.events.DiscoveryEvent;
	import org.apache.ignite.failure.FailureContext;
	import org.apache.ignite.internal.GridKernalContext;
	import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
	import org.apache.ignite.internal.IgniteInterruptedCheckedException;
	import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
	import org.apache.ignite.internal.events.DiscoveryCustomEvent;
	import org.apache.ignite.internal.managers.communication.GridMessageListener;
	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.managers.eventstorage.HighPriorityListener;
	import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
	import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
	import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
	import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
	import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
	import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
	import org.apache.ignite.internal.util.GridSpinBusyLock;
	import org.apache.ignite.internal.util.typedef.internal.U;
	import org.apache.ignite.internal.util.worker.GridWorker;
	import org.apache.ignite.thread.IgniteThread;
	import org.jetbrains.annotations.NotNull;
	import org.jetbrains.annotations.Nullable;

	import static org.apache.ignite.IgniteSystemProperties.IGNITE_LONG_OPERATIONS_DUMP_TIMEOUT_LIMIT;
	import static org.apache.ignite.IgniteSystemProperties.getLong;
	import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
	import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
	import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
	import static org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
	import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
	import static org.apache.ignite.internal.GridTopic.TOPIC_SERVICES;
	import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT;

	/**
	* Service deployment manager.
	*
	* @see ServiceDeploymentTask
	* @see ServiceDeploymentActions
	*/
	public class ServiceDeploymentManager {
	/** Busy lock. */
	private final GridSpinBusyLock busyLock = new GridSpinBusyLock();

	/** Services discovery messages listener. */
	private final DiscoveryEventListener discoLsnr = new ServiceDiscoveryListener();

	/** Services communication messages listener. */
	private final GridMessageListener commLsnr = new ServiceCommunicationListener();

	/** Services deployments tasks. */
	private final Map<ServiceDeploymentProcessId, ServiceDeploymentTask> tasks = new ConcurrentHashMap<>();

	/** Discovery events received while cluster state transition was in progress. */
	private final List<PendingEventHolder> pendingEvts = new ArrayList<>();

	/** Topology version of latest deployment task's event. */
	private final AtomicReference<AffinityTopologyVersion> readyTopVer =
	new AtomicReference<>(AffinityTopologyVersion.NONE);

	/** Kernal context. */
	private final GridKernalContext ctx;

	/** Logger. */
	private final IgniteLogger log;

	/** Deployment worker. */
	private final ServicesDeploymentWorker depWorker;

	/** Default dump operation limit. */
	private final long dfltDumpTimeoutLimit;

	/**
	* @param ctx Grid kernal context.
	*/
	ServiceDeploymentManager(@NotNull GridKernalContext ctx) {
	this.ctx = ctx;

	log = ctx.log(getClass());

	ctx.event().addDiscoveryEventListener(discoLsnr,
	EVT_NODE_JOINED, EVT_NODE_LEFT, EVT_NODE_FAILED, EVT_DISCOVERY_CUSTOM_EVT);

	ctx.io().addMessageListener(TOPIC_SERVICES, commLsnr);

	depWorker = new ServicesDeploymentWorker();

	long limit = getLong(IGNITE_LONG_OPERATIONS_DUMP_TIMEOUT_LIMIT, 0);

	dfltDumpTimeoutLimit = limit <= 0 ? 30 * 60_000 : limit;
	}

	/**
	* Starts processing of services deployments tasks.
	*/
	void startProcessing() {
	assert depWorker.runner() == null : "Method shouldn't be called twice during lifecycle;";

	new IgniteThread(ctx.igniteInstanceName(), "services-deployment-worker", depWorker).start();
	}

	/**
	* Stops processing of services deployments tasks.
	*
	* @param stopErr Cause error of deployment manager stop.
	*/
	void stopProcessing(IgniteCheckedException stopErr) {
	busyLock.block();

	try {
	ctx.event().removeDiscoveryEventListener(discoLsnr);

	ctx.io().removeMessageListener(commLsnr);

	U.cancel(depWorker);

	U.join(depWorker, log);

	depWorker.tasksQueue.clear();

	pendingEvts.clear();

	tasks.values().forEach(t -> t.completeError(stopErr));

	tasks.clear();
	}
	finally {
	busyLock.unblock();
	}
	}

	/**
	* @return Ready topology version.
	*/
	public AffinityTopologyVersion readyTopologyVersion() {
	return readyTopVer.get();
	}

	/**
	* Special handler for local discovery events for which the regular events are not generated, e.g. local join and
	* client reconnect events.
	*
	* @param evt Discovery event.
	* @param discoCache Discovery cache.
	* @param depActions Service deployment actions.
	*/
	void onLocalJoin(DiscoveryEvent evt, DiscoCache discoCache, ServiceDeploymentActions depActions) {
	checkClusterStateAndAddTask(evt, discoCache, depActions);
	}

	/**
	* Invokes {@link GridWorker#blockingSectionBegin()} for service deployment worker.
	* <p/>
	* Should be called from service deployment worker thread.
	*/
	void deployerBlockingSectionBegin() {
	assert depWorker != null && Thread.currentThread() == depWorker.runner();

	depWorker.blockingSectionBegin();
	}

	/**
	* Invokes {@link GridWorker#blockingSectionEnd()} for service deployment worker.
	* <p/>
	* Should be called from service deployment worker thread.
	*/
	void deployerBlockingSectionEnd() {
	assert depWorker != null && Thread.currentThread() == depWorker.runner();

	depWorker.blockingSectionEnd();
	}

	/**
	* Checks cluster state and handles given event.
	* <pre>
	* - if cluster is active, then adds event in deployment queue;
	* - if cluster state in transition, them adds to pending events;
	* - if cluster is inactive, then ignore event;
	* </pre>
	* <b>Should be called from discovery thread.</b>
	*
	* @param evt Discovery event.
	* @param discoCache Discovery cache.
	* @param depActions Services deployment actions.
	*/
	private void checkClusterStateAndAddTask(@NotNull DiscoveryEvent evt, @NotNull DiscoCache discoCache,
	@Nullable ServiceDeploymentActions depActions) {
	if (discoCache.state().transition())
	pendingEvts.add(new PendingEventHolder(evt, discoCache.version(), depActions));
	else if (discoCache.state().active())
	addTask(evt, discoCache.version(), depActions);
	else if (log.isDebugEnabled())
	log.debug("Ignore event, cluster is inactive, evt=" + evt);
	}

	/**
	* Adds deployment task with given deployment process id.
	* </p>
	* <b>Should be called from discovery thread.</b>
	*
	* @param evt Discovery event.
	* @param topVer Topology version.
	* @param depActions Services deployment actions.
	*/
	private void addTask(@NotNull DiscoveryEvent evt, @NotNull AffinityTopologyVersion topVer,
	@Nullable ServiceDeploymentActions depActions) {
	final ServiceDeploymentProcessId depId = deploymentId(evt, topVer);

	ServiceDeploymentTask task = tasks.computeIfAbsent(depId,
	t -> new ServiceDeploymentTask(ctx, depId));

	if (!task.onEnqueued()) {
	if (log.isDebugEnabled()) {
	log.debug("Service deployment process hasn't been started for discovery event, because of " +
	"a task with the same deployment process id is already added (possible cause is message's" +
	" double delivering), evt=" + evt);
	}

	return;
	}

	assert task.event() == null && task.topologyVersion() == null;

	task.onEvent(evt, topVer, depActions);

	depWorker.tasksQueue.add(task);
	}

	/**
	* Creates service deployment process id.
	*
	* @param evt Discovery event.
	* @param topVer Topology version.
	* @return Services deployment process id.
	*/
	private ServiceDeploymentProcessId deploymentId(@NotNull DiscoveryEvent evt,
	@NotNull AffinityTopologyVersion topVer) {
	return evt instanceof DiscoveryCustomEvent ?
	new ServiceDeploymentProcessId(((DiscoveryCustomEvent)evt).customMessage().id()) :
	new ServiceDeploymentProcessId(topVer);
	}

	/**
	* Clones some instances of {@link DiscoveryCustomEvent} to capture necessary data, to avoid custom messages's
	* nullifying by {@link GridDhtPartitionsExchangeFuture#onDone}.
	*
	* @param evt Discovery event.
	* @return Discovery event to process.
	*/
	private DiscoveryCustomEvent copyIfNeeded(@NotNull DiscoveryCustomEvent evt) {
	DiscoveryCustomMessage msg = evt.customMessage();

	assert msg != null : "DiscoveryCustomMessage has been nullified concurrently, evt=" + evt;

	if (msg instanceof ServiceChangeBatchRequest)
	return evt;

	DiscoveryCustomEvent cp = new DiscoveryCustomEvent();

	cp.node(evt.node());
	cp.customMessage(msg);
	cp.eventNode(evt.eventNode());
	cp.affinityTopologyVersion(evt.affinityTopologyVersion());

	return cp;
	}

	/**
	* Services discovery messages high priority listener.
	* <p/>
	* The listener should be notified earlier then PME's listener because of a custom message of {@link
	* DiscoveryCustomEvent} may be nullified in PME before the listener will be able to capture it.
	*/
	private class ServiceDiscoveryListener implements DiscoveryEventListener, HighPriorityListener {
	/** {@inheritDoc} */
	@Override public void onEvent(final DiscoveryEvent evt, final DiscoCache discoCache) {
	if (!enterBusy())
	return;

	try {
	final UUID snd = evt.eventNode().id();
	final int evtType = evt.type();

	assert snd != null : "Event's node id shouldn't be null.";
	assert evtType == EVT_NODE_JOINED \|\| evtType == EVT_NODE_LEFT \|\| evtType == EVT_NODE_FAILED
	\|\| evtType == EVT_DISCOVERY_CUSTOM_EVT : "Unexpected event was received, evt=" + evt;

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

	if (msg instanceof ChangeGlobalStateFinishMessage) {
	ChangeGlobalStateFinishMessage msg0 = (ChangeGlobalStateFinishMessage)msg;

	if (msg0.clusterActive())
	pendingEvts.forEach(t -> addTask(t.evt, t.topVer, t.depActions));
	else if (log.isDebugEnabled())
	pendingEvts.forEach(t -> log.debug("Ignore event, cluster is inactive: " + t.evt));

	pendingEvts.clear();
	}
	else {
	if (msg instanceof ServiceClusterDeploymentResultBatch) {
	ServiceClusterDeploymentResultBatch msg0 = (ServiceClusterDeploymentResultBatch)msg;

	if (log.isDebugEnabled()) {
	log.debug("Received services full deployments message : " +
	"[locId=" + ctx.localNodeId() + ", snd=" + snd + ", msg=" + msg0 + ']');
	}

	ServiceDeploymentProcessId depId = msg0.deploymentId();

	assert depId != null;

	ServiceDeploymentTask task = tasks.get(depId);

	if (task != null) // May be null in case of double delivering
	task.onReceiveFullDeploymentsMessage(msg0);
	}
	else if (msg instanceof CacheAffinityChangeMessage)
	addTask(copyIfNeeded((DiscoveryCustomEvent)evt), discoCache.version(), null);
	else {
	ServiceDeploymentActions depActions = null;

	if (msg instanceof ChangeGlobalStateMessage)
	depActions = ((ChangeGlobalStateMessage)msg).servicesDeploymentActions();
	else if (msg instanceof ServiceChangeBatchRequest) {
	depActions = ((ServiceChangeBatchRequest)msg)
	.servicesDeploymentActions();
	}
	else if (msg instanceof DynamicCacheChangeBatch)
	depActions = ((DynamicCacheChangeBatch)msg).servicesDeploymentActions();

	if (depActions != null)
	addTask(copyIfNeeded((DiscoveryCustomEvent)evt), discoCache.version(), depActions);
	}
	}
	}
	else {
	if (evtType == EVT_NODE_LEFT \|\| evtType == EVT_NODE_FAILED)
	tasks.values().forEach(t -> t.onNodeLeft(snd));

	checkClusterStateAndAddTask(evt, discoCache, null);
	}
	}
	finally {
	leaveBusy();
	}
	}

	/** {@inheritDoc} */
	@Override public int order() {
	return 0;
	}
	}

	/**
	* Pending event's holder.
	*/
	private static class PendingEventHolder {
	/** Discovery event. */
	private DiscoveryEvent evt;

	/** Topology version. */
	private AffinityTopologyVersion topVer;

	/** Services deployemnt actions. */
	private ServiceDeploymentActions depActions;

	/**
	* @param evt Discovery event.
	* @param topVer Topology version.
	* @param depActions Services deployment actions.
	*/
	private PendingEventHolder(DiscoveryEvent evt,
	AffinityTopologyVersion topVer, ServiceDeploymentActions depActions) {
	this.evt = evt;
	this.topVer = topVer;
	this.depActions = depActions;
	}
	}

	/**
	* Services messages communication listener.
	*/
	private class ServiceCommunicationListener implements GridMessageListener {
	/** {@inheritDoc} */
	@Override public void onMessage(UUID nodeId, Object msg, byte plc) {
	if (!enterBusy())
	return;

	try {
	if (msg instanceof ServiceSingleNodeDeploymentResultBatch) {
	ServiceSingleNodeDeploymentResultBatch msg0 = (ServiceSingleNodeDeploymentResultBatch)msg;

	if (log.isDebugEnabled()) {
	log.debug("Received services single deployments message : " +
	"[locId=" + ctx.localNodeId() + ", snd=" + nodeId + ", msg=" + msg0 + ']');
	}

	tasks.computeIfAbsent(msg0.deploymentId(),
	t -> new ServiceDeploymentTask(ctx, msg0.deploymentId()))
	.onReceiveSingleDeploymentsMessage(nodeId, msg0);
	}
	}
	finally {
	leaveBusy();
	}
	}
	}

	/**
	* Services deployment worker.
	*/
	private class ServicesDeploymentWorker extends GridWorker {
	/** Queue to process. */
	private final LinkedBlockingQueue<ServiceDeploymentTask> tasksQueue = new LinkedBlockingQueue<>();

	/** {@inheritDoc} */
	private ServicesDeploymentWorker() {
	super(ctx.igniteInstanceName(), "services-deployment-worker",
	ServiceDeploymentManager.this.log, ctx.workersRegistry());
	}

	/** {@inheritDoc} */
	@Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException {
	Throwable err = null;

	try {
	ServiceDeploymentTask task;

	while (!isCancelled()) {
	onIdle();

	blockingSectionBegin();

	try {
	task = tasksQueue.take();
	}
	finally {
	blockingSectionEnd();
	}

	if (isCancelled())
	Thread.currentThread().interrupt();

	task.init();

	final long dumpTimeout = 2 * ctx.config().getNetworkTimeout();

	long dumpCnt = 0;
	long nextDumpTime = 0;

	while (true) {
	try {
	blockingSectionBegin();

	try {
	task.waitForComplete(dumpTimeout);
	}
	finally {
	blockingSectionEnd();
	}

	taskPostProcessing(task);

	break;
	}
	catch (IgniteFutureTimeoutCheckedException ignored) {
	if (isCancelled())
	return;

	if (nextDumpTime <= U.currentTimeMillis()) {
	log.warning("Failed to wait service deployment process or timeout had been" +
	" reached, timeout=" + dumpTimeout +
	(log.isDebugEnabled() ? ", task=" + task : ", taskDepId=" + task.deploymentId()));

	long nextTimeout = dumpTimeout * (2 + dumpCnt++);

	nextDumpTime = U.currentTimeMillis() + Math.min(nextTimeout, dfltDumpTimeoutLimit);
	}
	}
	catch (ClusterTopologyServerNotFoundException e) {
	U.error(log, e);

	taskPostProcessing(task);

	break;
	}
	}
	}
	}
	catch (InterruptedException \| IgniteInterruptedCheckedException e) {
	Thread.currentThread().interrupt();

	if (!isCancelled())
	err = e;
	}
	catch (Throwable t) {
	err = t;
	}
	finally {
	if (err == null && !isCancelled())
	err = new IllegalStateException("Worker " + name() + " is terminated unexpectedly.");

	if (err instanceof OutOfMemoryError)
	ctx.failure().process(new FailureContext(CRITICAL_ERROR, err));
	else if (err != null)
	ctx.failure().process(new FailureContext(SYSTEM_WORKER_TERMINATION, err));
	}
	}

	/**
	* Does additional actions after task's completion.
	*/
	private void taskPostProcessing(ServiceDeploymentTask task) {
	AffinityTopologyVersion readyVer = readyTopVer.get();

	readyTopVer.compareAndSet(readyVer, task.topologyVersion());

	tasks.remove(task.deploymentId());
	}
	}

	/**
	* Enters busy state.
	*
	* @return {@code true} if entered to busy state.
	*/
	private boolean enterBusy() {
	return busyLock.enterBusy();
	}

	/**
	* Leaves busy state.
	*/
	private void leaveBusy() {
	busyLock.leaveBusy();
	}
	}