ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [04/23] incubator-ignite git commit: Merge branch 'sprint-1' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-24
Date Tue, 27 Jan 2015 13:34:02 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
index 0000000,65a3250..a921346
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
@@@ -1,0 -1,1371 +1,1373 @@@
+ /*
+  * 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.task;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.compute.*;
+ import org.apache.ignite.events.*;
+ import org.apache.ignite.fs.*;
+ import org.apache.ignite.internal.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.marshaller.*;
+ import org.apache.ignite.resources.*;
+ import org.apache.ignite.internal.managers.deployment.*;
+ import org.apache.ignite.internal.processors.job.*;
+ import org.apache.ignite.internal.processors.timeout.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ import org.apache.ignite.internal.util.worker.*;
+ import org.jdk8.backport.*;
+ import org.jetbrains.annotations.*;
+ 
+ import java.io.*;
+ import java.util.*;
+ import java.util.concurrent.*;
+ import java.util.concurrent.atomic.*;
+ 
+ import static org.apache.ignite.compute.ComputeJobResultPolicy.*;
+ import static org.apache.ignite.events.IgniteEventType.*;
+ import static org.apache.ignite.internal.GridTopic.*;
+ import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
+ import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.*;
+ 
+ /**
+  * Grid task worker. Handles full task life cycle.
+  * @param <T> Task argument type.
+  * @param <R> Task return value type.
+  */
+ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject {
+     /** Split size threshold. */
+     private static final int SPLIT_WARN_THRESHOLD = 1000;
+ 
+     /** {@code True} for internal tasks. */
+     private boolean internal;
+ 
+     /** */
+     private enum State {
+         /** */
+         WAITING,
+ 
+         /** */
+         REDUCING,
+ 
+         /** */
+         REDUCED,
+ 
+         /** */
+         FINISHING
+     }
+ 
+     /** Static logger to avoid re-creation. */
+     private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>();
+ 
+     /** */
+     private final GridKernalContext ctx;
+ 
+     /** */
+     private final IgniteLogger log;
+ 
+     /** */
+     private final IgniteMarshaller marsh;
+ 
+     /** */
+     private final GridTaskSessionImpl ses;
+ 
+     /** */
+     private final GridTaskFutureImpl<R> fut;
+ 
+     /** */
+     private final T arg;
+ 
+     /** */
+     private final GridTaskEventListener evtLsnr;
+ 
+     /** */
+     private Map<IgniteUuid, GridJobResultImpl> jobRes;
+ 
+     /** */
+     private State state = State.WAITING;
+ 
+     /** */
+     private final GridDeployment dep;
+ 
+     /** Task class. */
+     private final Class<?> taskCls;
+ 
+     /** Optional subgrid. */
+     private final Map<GridTaskThreadContextKey, Object> thCtx;
+ 
+     /** */
+     private ComputeTask<T, R> task;
+ 
+     /** */
+     private final Queue<GridJobExecuteResponse> delayedRess = new ConcurrentLinkedDeque8<>();
+ 
+     /** */
+     private boolean continuous;
+ 
+     /** */
+     private final Object mux = new Object();
+ 
+     /** */
+     private boolean lockRespProc = true;
+ 
+     /** */
+     private final boolean resCache;
+ 
+     /** */
+     private final boolean noFailover;
+ 
+     /** */
+     private final UUID subjId;
+ 
+     /** Continuous mapper. */
+     private final ComputeTaskContinuousMapper mapper = new ComputeTaskContinuousMapper() {
+         /** {@inheritDoc} */
+         @Override public void send(ComputeJob job, ClusterNode node) throws IgniteCheckedException {
+             A.notNull(job, "job");
+             A.notNull(node, "node");
+ 
+             processMappedJobs(Collections.singletonMap(job, node));
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public void send(Map<? extends ComputeJob, ClusterNode> mappedJobs) throws IgniteCheckedException {
+             A.notNull(mappedJobs, "mappedJobs");
+ 
+             processMappedJobs(mappedJobs);
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public void send(ComputeJob job) throws IgniteCheckedException {
+             A.notNull(job, "job");
+ 
+             send(Collections.singleton(job));
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public void send(Collection<? extends ComputeJob> jobs) throws IgniteCheckedException {
+             A.notNull(jobs, "jobs");
+ 
+             if (jobs.isEmpty())
+                 throw new IgniteCheckedException("Empty jobs collection passed to send(...) method.");
+ 
+             ComputeLoadBalancer balancer = ctx.loadBalancing().getLoadBalancer(ses, getTaskTopology());
+ 
+             for (ComputeJob job : jobs) {
+                 if (job == null)
+                     throw new IgniteCheckedException("Null job passed to send(...) method.");
+ 
+                 processMappedJobs(Collections.singletonMap(job, balancer.getBalancedNode(job, null)));
+             }
+         }
+     };
+ 
+     /**
+      * @param ctx Kernal context.
+      * @param arg Task argument.
+      * @param ses Grid task session.
+      * @param fut Task future.
+      * @param taskCls Task class.
+      * @param task Task instance that might be null.
+      * @param dep Deployed task.
+      * @param evtLsnr Event listener.
+      * @param thCtx Thread-local context from task processor.
+      * @param subjId Subject ID.
+      */
+     GridTaskWorker(
+         GridKernalContext ctx,
+         @Nullable T arg,
+         GridTaskSessionImpl ses,
+         GridTaskFutureImpl<R> fut,
+         @Nullable Class<?> taskCls,
+         @Nullable ComputeTask<T, R> task,
+         GridDeployment dep,
+         GridTaskEventListener evtLsnr,
+         @Nullable Map<GridTaskThreadContextKey, Object> thCtx,
+         UUID subjId) {
+         super(ctx.config().getGridName(), "grid-task-worker", ctx.config().getGridLogger());
+ 
+         assert ses != null;
+         assert fut != null;
+         assert evtLsnr != null;
+         assert dep != null;
+ 
+         this.arg = arg;
+         this.ctx = ctx;
+         this.fut = fut;
+         this.ses = ses;
+         this.taskCls = taskCls;
+         this.task = task;
+         this.dep = dep;
+         this.evtLsnr = evtLsnr;
+         this.thCtx = thCtx;
+         this.subjId = subjId;
+ 
+         log = U.logger(ctx, logRef, this);
+ 
+         marsh = ctx.config().getMarshaller();
+ 
+         resCache = dep.annotation(taskCls, ComputeTaskNoResultCache.class) == null;
+ 
+         Boolean noFailover = getThreadContext(TC_NO_FAILOVER);
+ 
+         this.noFailover = noFailover != null ? noFailover : false;
+     }
+ 
+     /**
+      * Gets value from thread-local context.
+      *
+      * @param key Thread-local context key.
+      * @return Thread-local context value, if any.
+      */
+     @SuppressWarnings({"unchecked"})
+     @Nullable private <V> V getThreadContext(GridTaskThreadContextKey key) {
+         return thCtx == null ? null : (V)thCtx.get(key);
+     }
+ 
+     /**
+      * @return Task session ID.
+      */
+     IgniteUuid getTaskSessionId() {
+         return ses.getId();
+     }
+ 
+     /**
+      * @return Task session.
+      */
+     GridTaskSessionImpl getSession() {
+         return ses;
+     }
+ 
+     /**
+      * @return Task future.
+      */
+     GridTaskFutureImpl<R> getTaskFuture() {
+         return fut;
+     }
+ 
+     /**
+      * Gets property dep.
+      *
+      * @return Property dep.
+      */
+     GridDeployment getDeployment() {
+         return dep;
+     }
+ 
+     /**
+      * @return Grid task.
+      */
+     public ComputeTask<T, R> getTask() {
+         return task;
+     }
+ 
+     /**
+      * @param task Deployed task.
+      */
+     public void setTask(ComputeTask<T, R> task) {
+         this.task = task;
+     }
+ 
+     /**
+      * @return {@code True} if task is internal.
+      */
+     public boolean isInternal() {
+         return internal;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public IgniteUuid timeoutId() {
+         return ses.getId();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void onTimeout() {
+         synchronized (mux) {
+             if (state != State.WAITING)
+                 return;
+         }
+ 
+         U.warn(log, "Task has timed out: " + ses);
+ 
+         recordTaskEvent(EVT_TASK_TIMEDOUT, "Task has timed out.");
+ 
+         Throwable e = new ComputeTaskTimeoutException("Task timed out (check logs for error messages): " + ses);
+ 
+         finishTask(null, e);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public long endTime() {
+         return ses.getEndTime();
+     }
+ 
+     /**
+      * @param taskCls Task class.
+      * @return Task instance.
+      * @throws IgniteCheckedException Thrown in case of any instantiation error.
+      */
+     private ComputeTask<T, R> newTask(Class<? extends ComputeTask<T, R>> taskCls) throws IgniteCheckedException {
+         ComputeTask<T, R> task = dep.newInstance(taskCls);
+ 
+         if (task == null)
+             throw new IgniteCheckedException("Failed to instantiate task (is default constructor available?): " + taskCls);
+ 
+         return task;
+     }
+ 
+     /**
+      *
+      */
+     private void initializeSpis() {
+         ComputeTaskSpis spis = dep.annotation(taskCls, ComputeTaskSpis.class);
+ 
+         if (spis != null) {
+             ses.setLoadBalancingSpi(spis.loadBalancingSpi());
+             ses.setFailoverSpi(spis.failoverSpi());
+             ses.setCheckpointSpi(spis.checkpointSpi());
+         }
+     }
+ 
+     /**
+      * Maps this task's jobs to nodes and sends them out.
+      */
+     @SuppressWarnings({"unchecked"})
+     @Override protected void body() {
+         evtLsnr.onTaskStarted(this);
+ 
+         try {
+             // Use either user task or deployed one.
+             if (task == null) {
+                 assert taskCls != null;
+                 assert ComputeTask.class.isAssignableFrom(taskCls);
+ 
+                 try {
+                     task = newTask((Class<? extends ComputeTask<T, R>>)taskCls);
+                 }
+                 catch (IgniteCheckedException e) {
+                     // If cannot instantiate task, then assign internal flag based
+                     // on information available.
+                     internal = dep.internalTask(null, taskCls);
+ 
+                     recordTaskEvent(EVT_TASK_STARTED, "Task started.");
+ 
+                     throw e;
+                 }
+             }
+ 
+             internal = dep.internalTask(task, taskCls);
+ 
+             recordTaskEvent(EVT_TASK_STARTED, "Task started.");
+ 
+             initializeSpis();
+ 
+             ses.setClassLoader(dep.classLoader());
+ 
+             final List<ClusterNode> shuffledNodes = getTaskTopology();
+ 
+             // Load balancer.
+             ComputeLoadBalancer balancer = ctx.loadBalancing().getLoadBalancer(ses, shuffledNodes);
+ 
+             continuous = ctx.resource().isAnnotationPresent(dep, task, IgniteTaskContinuousMapperResource.class);
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Injected task resources [continuous=" + continuous + ']');
+ 
+             // Inject resources.
+             ctx.resource().inject(dep, task, ses, balancer, mapper);
+ 
+             Map<? extends ComputeJob, ClusterNode> mappedJobs = U.wrapThreadLoader(dep.classLoader(),
+                 new Callable<Map<? extends ComputeJob, ClusterNode>>() {
+                     @Override public Map<? extends ComputeJob, ClusterNode> call() throws IgniteCheckedException {
+                         return task.map(shuffledNodes, arg);
+                     }
+                 });
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Mapped task jobs to nodes [jobCnt=" + (mappedJobs != null ? mappedJobs.size() : 0) +
+                     ", mappedJobs=" + mappedJobs + ", ses=" + ses + ']');
+ 
+             if (F.isEmpty(mappedJobs)) {
+                 synchronized (mux) {
+                     // Check if some jobs are sent from continuous mapper.
+                     if (F.isEmpty(jobRes))
+                         throw new IgniteCheckedException("Task map operation produced no mapped jobs: " + ses);
+                 }
+             }
+             else
+                 processMappedJobs(mappedJobs);
+ 
+             synchronized (mux) {
+                 lockRespProc = false;
+             }
+ 
+             processDelayedResponses();
+         }
+         catch (ClusterGroupEmptyException e) {
+             U.warn(log, "Failed to map task jobs to nodes (topology projection is empty): " + ses);
+ 
+             finishTask(null, e);
+         }
+         catch (IgniteCheckedException e) {
+             if (!fut.isCancelled()) {
+                 U.error(log, "Failed to map task jobs to nodes: " + ses, e);
+ 
+                 finishTask(null, e);
+             }
+             else if (log.isDebugEnabled())
+                 log.debug("Failed to map task jobs to nodes due to task cancellation: " + ses);
+         }
+         // Catch throwable to protect against bad user code.
+         catch (Throwable e) {
+             String errMsg = "Failed to map task jobs to nodes due to undeclared user exception" +
+                 " [cause=" + e.getMessage() + ", ses=" + ses + "]";
+ 
+             U.error(log, errMsg, e);
+ 
+             finishTask(null, new ComputeUserUndeclaredException(errMsg, e));
+         }
+     }
+ 
+     /**
+      * @param jobs Map of jobs.
+      * @throws IgniteCheckedException Thrown in case of any error.
+      */
+     private void processMappedJobs(Map<? extends ComputeJob, ClusterNode> jobs) throws IgniteCheckedException {
+         if (F.isEmpty(jobs))
+             return;
+ 
+         Collection<GridJobResultImpl> jobResList = new ArrayList<>(jobs.size());
+ 
+         Collection<ComputeJobSibling> sibs = new ArrayList<>(jobs.size());
+ 
+         // Map jobs to nodes for computation.
+         for (Map.Entry<? extends ComputeJob, ClusterNode> mappedJob : jobs.entrySet()) {
+             ComputeJob job = mappedJob.getKey();
+             ClusterNode node = mappedJob.getValue();
+ 
+             if (job == null)
+                 throw new IgniteCheckedException("Job can not be null [mappedJob=" + mappedJob + ", ses=" + ses + ']');
+ 
+             if (node == null)
+                 throw new IgniteCheckedException("Node can not be null [mappedJob=" + mappedJob + ", ses=" + ses + ']');
+ 
+             IgniteUuid jobId = IgniteUuid.fromUuid(node.id());
+ 
+             GridJobSiblingImpl sib = new GridJobSiblingImpl(ses.getId(), jobId, node.id(), ctx);
+ 
+             jobResList.add(new GridJobResultImpl(job, jobId, node, sib));
+ 
+             // Do not add siblings if result cache is disabled.
+             if (resCache)
+                 sibs.add(sib);
+ 
+             recordJobEvent(EVT_JOB_MAPPED, jobId, node, "Job got mapped.");
+         }
+ 
+         synchronized (mux) {
+             if (state != State.WAITING)
+                 throw new IgniteCheckedException("Task is not in waiting state [state=" + state + ", ses=" + ses + ']');
+ 
+             // Do not add siblings if result cache is disabled.
+             if (resCache)
+                 ses.addJobSiblings(sibs);
+ 
+             if (jobRes == null)
+                 jobRes = new HashMap<>();
+ 
+             // Populate all remote mappedJobs into map, before mappedJobs are sent.
+             // This is done to avoid race condition when we start
+             // getting results while still sending out references.
+             for (GridJobResultImpl res : jobResList) {
+                 if (jobRes.put(res.getJobContext().getJobId(), res) != null)
+                     throw new IgniteCheckedException("Duplicate job ID for remote job found: " + res.getJobContext().getJobId());
+ 
+                 res.setOccupied(true);
+ 
+                 if (resCache && jobRes.size() > ctx.discovery().size() && jobRes.size() % SPLIT_WARN_THRESHOLD == 0)
+                     LT.warn(log, null, "Number of jobs in task is too large for task: " + ses.getTaskName() +
+                         ". Consider reducing number of jobs or disabling job result cache with " +
+                         "@GridComputeTaskNoResultCache annotation.");
+             }
+         }
+ 
+         // Set mapped flag.
+         ses.onMapped();
+ 
+         // Send out all remote mappedJobs.
+         for (GridJobResultImpl res : jobResList) {
+             evtLsnr.onJobSend(this, res.getSibling());
+ 
+             try {
+                 sendRequest(res);
+             }
+             finally {
+                 // Open job for processing results.
+                 synchronized (mux) {
+                     res.setOccupied(false);
+                 }
+             }
+         }
+ 
+         processDelayedResponses();
+     }
+ 
+     /**
+      * @return Topology for this task.
+      * @throws IgniteCheckedException Thrown in case of any error.
+      */
+     private List<ClusterNode> getTaskTopology() throws IgniteCheckedException {
+         Collection<UUID> top = ses.getTopology();
+ 
+         Collection<? extends ClusterNode> subgrid = top != null ? ctx.discovery().nodes(top) : ctx.discovery().allNodes();
+ 
+         int size = subgrid.size();
+ 
+         if (size == 0)
+             throw new ClusterGroupEmptyException("Topology projection is empty.");
+ 
+         List<ClusterNode> shuffledNodes = new ArrayList<>(size);
+ 
+         for (ClusterNode node : subgrid)
+             shuffledNodes.add(node);
+ 
+         if (shuffledNodes.size() > 1)
+             // Shuffle nodes prior to giving them to user.
+             Collections.shuffle(shuffledNodes);
+ 
+         // Load balancer.
+         return shuffledNodes;
+     }
+ 
+     /**
+      *
+      */
+     private void processDelayedResponses() {
+         GridJobExecuteResponse res = delayedRess.poll();
+ 
+         if (res != null)
+             onResponse(res);
+     }
+ 
+     /**
+      * @param msg Job execution response.
+      */
+     void onResponse(GridJobExecuteResponse msg) {
+         assert msg != null;
+ 
+         if (fut.isDone()) {
+             if (log.isDebugEnabled())
+                 log.debug("Ignoring job response since task has finished: " + msg);
+ 
+             return;
+         }
+ 
+         GridJobExecuteResponse res = msg;
+ 
+         while (res != null) {
+             GridJobResultImpl jobRes = null;
+ 
+             // Flag indicating whether occupied flag for
+             // job response was changed in this method apply.
+             boolean selfOccupied = false;
+ 
+             try {
+                 synchronized (mux) {
+                     // If task is not waiting for responses,
+                     // then there is no point to proceed.
+                     if (state != State.WAITING) {
+                         if (log.isDebugEnabled())
+                             log.debug("Ignoring response since task is already reducing or finishing [res=" + res +
+                                 ", job=" + ses + ", state=" + state + ']');
+ 
+                         return;
+                     }
+ 
+                     jobRes = this.jobRes.get(res.getJobId());
+ 
+                     if (jobRes == null) {
+                         if (log.isDebugEnabled())
+                             U.warn(log, "Received response for unknown child job (was job presumed failed?): " + res);
+ 
+                         return;
+                     }
+ 
+                     // Only process 1st response and ignore following ones. This scenario
+                     // is possible if node has left topology and and fake failure response
+                     // was created from discovery listener and when sending request failed.
+                     if (jobRes.hasResponse()) {
+                         if (log.isDebugEnabled())
+                             log.debug("Received redundant response for a job (will ignore): " + res);
+ 
+                         return;
+                     }
+ 
+                     if (!jobRes.getNode().id().equals(res.getNodeId())) {
+                         if (log.isDebugEnabled())
+                             log.debug("Ignoring stale response as job was already resent to other node [res=" + res +
+                                 ", jobRes=" + jobRes + ']');
+ 
+                         // Prevent processing 2 responses for the same job simultaneously.
+                         jobRes.setOccupied(true);
+ 
+                         selfOccupied = true;
+ 
+                         // We can not return here because there can be more delayed messages in the queue.
+                         continue;
+                     }
+ 
+                     if (jobRes.isOccupied()) {
+                         if (log.isDebugEnabled())
+                             log.debug("Adding response to delayed queue (job is either being sent or processing " +
+                                 "another response): " + res);
+ 
+                         delayedRess.offer(res);
+ 
+                         return;
+                     }
+ 
+                     if (lockRespProc) {
+                         delayedRess.offer(res);
+ 
+                         return;
+                     }
+ 
+                     lockRespProc = true;
+ 
+                     selfOccupied = true;
+ 
+                     // Prevent processing 2 responses for the same job simultaneously.
+                     jobRes.setOccupied(true);
+ 
+                     // We don't keep reference to job if results are not cached.
+                     if (!resCache)
+                         this.jobRes.remove(res.getJobId());
+                 }
+ 
+                 if (res.getFakeException() != null)
+                     jobRes.onResponse(null, res.getFakeException(), null, false);
+                 else {
+                     ClassLoader clsLdr = dep.classLoader();
+ 
+                     try {
+                         boolean loc = ctx.localNodeId().equals(res.getNodeId()) && !ctx.config().isMarshalLocalJobs();
+ 
+                         Object res0 = loc ? res.getJobResult() : marsh.unmarshal(res.getJobResultBytes(), clsLdr);
+ 
+                         IgniteCheckedException ex = loc ? res.getException() :
+                             marsh.<IgniteCheckedException>unmarshal(res.getExceptionBytes(), clsLdr);
+ 
+                         Map<Object, Object> attrs = loc ? res.getJobAttributes() :
+                             marsh.<Map<Object, Object>>unmarshal(res.getJobAttributesBytes(), clsLdr);
+ 
+                         jobRes.onResponse(res0, ex, attrs, res.isCancelled());
+ 
+                         if (loc)
+                             ctx.resource().invokeAnnotated(dep, jobRes.getJob(), ComputeJobAfterSend.class);
+                     }
+                     catch (IgniteCheckedException e) {
+                         U.error(log, "Error deserializing job response: " + res, e);
+ 
+                         finishTask(null, e);
+                     }
+                 }
+ 
+                 List<ComputeJobResult> results;
+ 
+                 if (!resCache)
+                     results = Collections.emptyList();
+                 else {
+                     synchronized (mux) {
+                         results = getRemoteResults();
+                     }
+                 }
+ 
+                 ComputeJobResultPolicy plc = result(jobRes, results);
+ 
+                 if (plc == null) {
+                     String errMsg = "Failed to obtain remote job result policy for result from GridComputeTask.result(..) " +
+                         "method that returned null (will fail the whole task): " + jobRes;
+ 
+                     finishTask(null, new IgniteCheckedException(errMsg));
+ 
+                     return;
+                 }
+ 
+                 synchronized (mux) {
+                     // If task is not waiting for responses,
+                     // then there is no point to proceed.
+                     if (state != State.WAITING) {
+                         if (log.isDebugEnabled())
+                             log.debug("Ignoring GridComputeTask.result(..) value since task is already reducing or" +
+                                 "finishing [res=" + res + ", job=" + ses + ", state=" + state + ']');
+ 
+                         return;
+                     }
+ 
+                     switch (plc) {
+                         // Start reducing all results received so far.
+                         case REDUCE: {
+                             state = State.REDUCING;
+ 
+                             break;
+                         }
+ 
+                         // Keep waiting if there are more responses to come,
+                         // otherwise, reduce.
+                         case WAIT: {
+                             assert results.size() <= this.jobRes.size();
+ 
+                             // If there are more results to wait for.
+                             // If result cache is disabled, then we reduce
+                             // when both collections are empty.
+                             if (results.size() == this.jobRes.size()) {
+                                 plc = ComputeJobResultPolicy.REDUCE;
+ 
+                                 // All results are received, proceed to reduce method.
+                                 state = State.REDUCING;
+                             }
+ 
+                             break;
+                         }
+ 
+                         case FAILOVER: {
+                             if (!failover(res, jobRes, getTaskTopology()))
+                                 plc = null;
+ 
+                             break;
+                         }
+                     }
+                 }
+ 
+                 // Outside of synchronization.
+                 if (plc != null) {
+                     // Handle failover.
+                     if (plc == FAILOVER)
+                         sendFailoverRequest(jobRes);
+                     else {
+                         evtLsnr.onJobFinished(this, jobRes.getSibling());
+ 
+                         if (plc == ComputeJobResultPolicy.REDUCE)
+                             reduce(results);
+                     }
+                 }
+             }
+             catch (IgniteCheckedException e) {
+                 U.error(log, "Failed to obtain topology [ses=" + ses + ", err=" + e + ']', e);
+ 
+                 finishTask(null, e);
+             }
+             finally {
+                 // Open up job for processing responses.
+                 // Only unset occupied flag, if it was
+                 // set in this method.
+                 if (selfOccupied) {
+                     assert jobRes != null;
+ 
+                     synchronized (mux) {
+                         jobRes.setOccupied(false);
+ 
+                         lockRespProc = false;
+                     }
+ 
+                     // Process delayed responses if there are any.
+                     res = delayedRess.poll();
+                 }
+             }
+         }
+     }
+ 
+     /**
+      * @param jobRes Job result.
+      * @param results Existing job results.
+      * @return Job result policy.
+      */
+     @SuppressWarnings({"CatchGenericClass"})
+     @Nullable private ComputeJobResultPolicy result(final ComputeJobResult jobRes, final List<ComputeJobResult> results) {
+         assert !Thread.holdsLock(mux);
+ 
+         return U.wrapThreadLoader(dep.classLoader(), new CO<ComputeJobResultPolicy>() {
+             @Nullable @Override public ComputeJobResultPolicy apply() {
+                 try {
+                     // Obtain job result policy.
+                     ComputeJobResultPolicy plc = null;
+ 
+                     try {
+                         plc = task.result(jobRes, results);
+ 
+                         if (plc == FAILOVER && noFailover) {
+                             IgniteCheckedException e = jobRes.getException();
+ 
+                             if (e != null)
+                                 throw e;
+ 
+                             plc = WAIT;
+                         }
+                     }
+                     finally {
+                         recordJobEvent(EVT_JOB_RESULTED, jobRes.getJobContext().getJobId(),
+                             jobRes.getNode(), "Job got resulted with: " + plc);
+                     }
+ 
+                     if (log.isDebugEnabled())
+                         log.debug("Obtained job result policy [policy=" + plc + ", ses=" + ses + ']');
+ 
+                     return plc;
+                 }
+                 catch (IgniteCheckedException e) {
+                     if (X.hasCause(e, GridInternalException.class) ||
+                         X.hasCause(e, IgniteFsOutOfSpaceException.class)) {
+                         // Print internal exceptions only if debug is enabled.
+                         if (log.isDebugEnabled())
+                             U.error(log, "Failed to obtain remote job result policy for result from " +
+                                 "GridComputeTask.result(..) method (will fail the whole task): " + jobRes, e);
+                     }
+                     else if (X.hasCause(e, ComputeJobFailoverException.class)) {
+                         IgniteCheckedException e0 = new IgniteCheckedException(" Job was not failed over because " +
+                             "GridComputeJobResultPolicy.FAILOVER was not returned from " +
+                             "GridTask.result(...) method for job result with GridComputeJobFailoverException.", e);
+ 
+                         finishTask(null, e0);
+ 
+                         return null;
+                     }
+                     else
+                         U.error(log, "Failed to obtain remote job result policy for result from " +
+                             "GridComputeTask.result(..) method (will fail the whole task): " + jobRes, e);
+ 
+                     finishTask(null, e);
+ 
+                     return null;
+                 }
+                 catch (IgniteException e) {
+                     if (X.hasCause(e, GridInternalException.class) ||
+                         X.hasCause(e, IgniteFsOutOfSpaceException.class)) {
+                         // Print internal exceptions only if debug is enabled.
+                         if (log.isDebugEnabled())
+                             U.error(log, "Failed to obtain remote job result policy for result from " +
+                                 "GridComputeTask.result(..) method (will fail the whole task): " + jobRes, e);
+                     }
+                     else if (X.hasCause(e, ComputeJobFailoverException.class)) {
+                         IgniteCheckedException e0 = new IgniteCheckedException(" Job was not failed over because " +
+                             "GridComputeJobResultPolicy.FAILOVER was not returned from " +
+                             "GridTask.result(...) method for job result with GridComputeJobFailoverException.", e);
+ 
+                         finishTask(null, e0);
+ 
+                         return null;
+                     }
+                     else
+                         U.error(log, "Failed to obtain remote job result policy for result from" +
+                             "GridComputeTask.result(..) method (will fail the whole task): " + jobRes, e);
+ 
+                     finishTask(null, e);
+ 
+                     return null;
+                 }
+                 catch (Throwable e) {
+                     String errMsg = "Failed to obtain remote job result policy for result from" +
+                         "GridComputeTask.result(..) method due to undeclared user exception " +
+                         "(will fail the whole task): " + jobRes;
+ 
+                     U.error(log, errMsg, e);
+ 
+                     Throwable tmp = new ComputeUserUndeclaredException(errMsg, e);
+ 
+                     // Failed to successfully obtain result policy and
+                     // hence forced to fail the whole deployed task.
+                     finishTask(null, tmp);
+ 
+                     return null;
+                 }
+             }
+         });
+     }
+ 
+     /**
+      * @param results Job results.
+      */
+     private void reduce(final List<ComputeJobResult> results) {
+         R reduceRes = null;
+         Throwable userE = null;
+ 
+         try {
+             try {
+                 // Reduce results.
+                 reduceRes = U.wrapThreadLoader(dep.classLoader(), new Callable<R>() {
+                     @Nullable @Override public R call() throws IgniteCheckedException {
+                         return task.reduce(results);
+                     }
+                 });
+             }
+             finally {
+                 synchronized (mux) {
+                     assert state == State.REDUCING : "Invalid task state: " + state;
+ 
+                     state = State.REDUCED;
+                 }
+             }
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Reduced job responses [reduceRes=" + reduceRes + ", ses=" + ses + ']');
+ 
+             recordTaskEvent(EVT_TASK_REDUCED, "Task reduced.");
+         }
+         catch (ClusterTopologyException e) {
+             U.warn(log, "Failed to reduce job results for task (any nodes from task topology left grid?): " + task);
+ 
+             userE = e;
+         }
+         catch (IgniteCheckedException e) {
+             U.error(log, "Failed to reduce job results for task: " + task, e);
+ 
+             userE = e;
+         }
+         // Catch Throwable to protect against bad user code.
+         catch (Throwable e) {
+             String errMsg = "Failed to reduce job results due to undeclared user exception [task=" + task +
+                 ", err=" + e + ']';
+ 
+             U.error(log, errMsg, e);
+ 
+             userE = new ComputeUserUndeclaredException(errMsg ,e);
+         }
+         finally {
+             finishTask(reduceRes, userE);
+         }
+     }
+ 
+     /**
+      * @param res Execution response.
+      * @param jobRes Job result.
+      * @param top Topology.
+      * @return {@code True} if fail-over SPI returned a new node.
+      */
+     private boolean failover(GridJobExecuteResponse res, GridJobResultImpl jobRes, Collection<? extends ClusterNode> top) {
+         assert Thread.holdsLock(mux);
+ 
+         try {
+             ctx.resource().invokeAnnotated(dep, jobRes.getJob(), ComputeJobBeforeFailover.class);
+ 
+             // Map to a new node.
+             ClusterNode node = ctx.failover().failover(ses, jobRes, new ArrayList<>(top));
+ 
+             if (node == null) {
+                 String msg = "Failed to failover a job to another node (failover SPI returned null) [job=" +
+                     jobRes.getJob() + ", node=" + jobRes.getNode() + ']';
+ 
+                 if (log.isDebugEnabled())
+                     log.debug(msg);
+ 
+                 Throwable e = new ClusterTopologyException(msg, jobRes.getException());
+ 
+                 finishTask(null, e);
+ 
+                 return false;
+             }
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Resolved job failover [newNode=" + node + ", oldNode=" + jobRes.getNode() +
+                     ", job=" + jobRes.getJob() + ", resMsg=" + res + ']');
+ 
+             jobRes.setNode(node);
+             jobRes.resetResponse();
+ 
+             if (!resCache) {
+                 synchronized (mux) {
+                     // Store result back in map before sending.
+                     this.jobRes.put(res.getJobId(), jobRes);
+                 }
+             }
+ 
+             return true;
+         }
+         // Catch Throwable to protect against bad user code.
+         catch (Throwable e) {
+             String errMsg = "Failed to failover job due to undeclared user exception [job=" +
+                 jobRes.getJob() + ", err=" + e + ']';
+ 
+             U.error(log, errMsg, e);
+ 
+             finishTask(null, new ComputeUserUndeclaredException(errMsg, e));
+ 
+             return false;
+         }
+     }
+ 
+     /**
+      * @param jobRes Job result.
+      */
+     private void sendFailoverRequest(GridJobResultImpl jobRes) {
+         // Internal failover notification.
+         evtLsnr.onJobFailover(this, jobRes.getSibling(), jobRes.getNode().id());
+ 
+         long timeout = ses.getEndTime() - U.currentTimeMillis();
+ 
+         if (timeout > 0) {
+             recordJobEvent(EVT_JOB_FAILED_OVER, jobRes.getJobContext().getJobId(),
+                 jobRes.getNode(), "Job failed over.");
+ 
+             // Send new reference to remote nodes for execution.
+             sendRequest(jobRes);
+         }
+         else
+             // Don't apply 'finishTask(..)' here as it will
+             // be called from 'onTimeout(..)' callback.
+             U.warn(log, "Failed to fail-over job due to task timeout: " + jobRes);
+     }
+ 
+     /**
+      * Interrupts child jobs on remote nodes.
+      */
+     private void cancelChildren() {
+         Collection<GridJobResultImpl> doomed = new LinkedList<>();
+ 
+         synchronized (mux) {
+             // Only interrupt unfinished jobs.
+             if (jobRes != null)
+                 for (GridJobResultImpl res : jobRes.values())
+                     if (!res.hasResponse())
+                         doomed.add(res);
+         }
+ 
+         // Send cancellation request to all unfinished children.
+         for (GridJobResultImpl res : doomed) {
+             UUID nodeId = res.getNode().id();
+ 
+             if (nodeId.equals(ctx.localNodeId()))
+                 // Cancel local jobs.
+                 ctx.job().cancelJob(ses.getId(), res.getJobContext().getJobId(), /*courtesy*/true);
+             else {
+                 try {
+                     ClusterNode node = ctx.discovery().node(nodeId);
+ 
+                     if (node != null)
+                         ctx.io().send(node,
+                             TOPIC_JOB_CANCEL,
+                             new GridJobCancelRequest(ses.getId(), res.getJobContext().getJobId(), /*courtesy*/true),
+                             PUBLIC_POOL);
+                 }
+                 catch (IgniteCheckedException e) {
+                     if (!isDeadNode(nodeId))
+                         U.error(log, "Failed to send cancel request to node (will ignore) [nodeId=" +
+                             nodeId + ", taskName=" + ses.getTaskName() +
+                             ", taskSesId=" + ses.getId() + ", jobSesId=" + res.getJobContext().getJobId() + ']', e);
+                 }
+             }
+         }
+     }
+ 
+     /**
+      * @param res Job result.
+      */
+     private void sendRequest(ComputeJobResult res) {
+         assert res != null;
+ 
+         GridJobExecuteRequest req = null;
+ 
+         ClusterNode node = res.getNode();
+ 
+         try {
+             ClusterNode curNode = ctx.discovery().node(node.id());
+ 
+             // Check if node exists prior to sending to avoid cases when a discovery
+             // listener notified about node leaving after topology resolution. Note
+             // that we make this check because we cannot count on exception being
+             // thrown in case of send failure.
+             if (curNode == null) {
+                 U.warn(log, "Failed to send job request because remote node left grid (if fail-over is enabled, " +
+                     "will attempt fail-over to another node) [node=" + node + ", taskName=" + ses.getTaskName() +
+                     ", taskSesId=" + ses.getId() + ", jobSesId=" + res.getJobContext().getJobId() + ']');
+ 
+                 ctx.resource().invokeAnnotated(dep, res.getJob(), ComputeJobAfterSend.class);
+ 
+                 GridJobExecuteResponse fakeRes = new GridJobExecuteResponse(node.id(), ses.getId(),
+                     res.getJobContext().getJobId(), null, null, null, null, null, null, false);
+ 
+                 fakeRes.setFakeException(new ClusterTopologyException("Failed to send job due to node failure: " + node));
+ 
+                 onResponse(fakeRes);
+             }
+             else {
+                 long timeout = ses.getEndTime() == Long.MAX_VALUE ? Long.MAX_VALUE :
+                     ses.getEndTime() - U.currentTimeMillis();
+ 
+                 if (timeout > 0) {
+                     boolean loc = node.id().equals(ctx.discovery().localNode().id()) &&
+                         !ctx.config().isMarshalLocalJobs();
+ 
+                     Map<Object, Object> sesAttrs = ses.isFullSupport() ? ses.getAttributes() : null;
+                     Map<? extends Serializable, ? extends Serializable> jobAttrs =
+                         (Map<? extends Serializable, ? extends Serializable>)res.getJobContext().getAttributes();
+ 
+                     boolean forceLocDep = internal || !ctx.deploy().enabled();
+ 
+                     req = new GridJobExecuteRequestV2(
+                             ses.getId(),
+                             res.getJobContext().getJobId(),
+                             ses.getTaskName(),
+                             ses.getUserVersion(),
+                             ses.getTaskClassName(),
+                             loc ? null : marsh.marshal(res.getJob()),
+                             loc ? res.getJob() : null,
+                             ses.getStartTime(),
+                             timeout,
+                             ses.getTopology(),
+                             loc ? null : marsh.marshal(ses.getJobSiblings()),
+                             loc ? ses.getJobSiblings() : null,
+                             loc ? null : marsh.marshal(sesAttrs),
+                             loc ? sesAttrs : null,
+                             loc ? null: marsh.marshal(jobAttrs),
+                             loc ? jobAttrs : null,
+                             ses.getCheckpointSpi(),
+                             dep.classLoaderId(),
+                             dep.deployMode(),
+                             continuous,
+                             dep.participants(),
+                             forceLocDep,
+                             ses.isFullSupport(),
+                             internal,
+                             subjId);
+ 
+                     if (loc)
+                         ctx.job().processJobExecuteRequest(ctx.discovery().localNode(), req);
+                     else {
+                         // Send job execution request.
+                         ctx.io().send(node, TOPIC_JOB, req, internal ? MANAGEMENT_POOL : PUBLIC_POOL);
+ 
+                         if (log.isDebugEnabled())
+                             log.debug("Sent job request [req=" + req + ", node=" + node + ']');
+                     }
+ 
+                     if (!loc)
+                         ctx.resource().invokeAnnotated(dep, res.getJob(), ComputeJobAfterSend.class);
+                 }
+                 else
+                     U.warn(log, "Job timed out prior to sending job execution request: " + res.getJob());
+             }
+         }
+         catch (IgniteCheckedException e) {
+             boolean deadNode = isDeadNode(res.getNode().id());
+ 
+             // Avoid stack trace if node has left grid.
+             if (deadNode)
+                 U.warn(log, "Failed to send job request because remote node left grid (if failover is enabled, " +
+                     "will attempt fail-over to another node) [node=" + node + ", taskName=" + ses.getTaskName() +
+                     ", taskSesId=" + ses.getId() + ", jobSesId=" + res.getJobContext().getJobId() + ']');
+             else
+                 U.error(log, "Failed to send job request: " + req, e);
+ 
+             GridJobExecuteResponse fakeRes = new GridJobExecuteResponse(node.id(), ses.getId(),
+                 res.getJobContext().getJobId(), null, null, null, null, null, null, false);
+ 
+             if (deadNode)
+                 fakeRes.setFakeException(new ClusterTopologyException("Failed to send job due to node failure: " +
+                     node, e));
+             else
+                 fakeRes.setFakeException(e);
+ 
+             onResponse(fakeRes);
+         }
+     }
+ 
+     /**
+      * @param nodeId Node ID.
+      */
+     void onNodeLeft(UUID nodeId) {
+         Collection<GridJobExecuteResponse> resList = null;
+ 
+         synchronized (mux) {
+             // First check if job cares about future responses.
+             if (state != State.WAITING)
+                 return;
+ 
+             if (jobRes != null) {
+                 for (GridJobResultImpl jr : jobRes.values()) {
+                     if (!jr.hasResponse() && jr.getNode().id().equals(nodeId)) {
+                         if (log.isDebugEnabled())
+                             log.debug("Creating fake response because node left grid [job=" + jr.getJob() +
+                                 ", nodeId=" + nodeId + ']');
+ 
+                         // Artificial response in case if a job is waiting for a response from
+                         // non-existent node.
+                         GridJobExecuteResponse fakeRes = new GridJobExecuteResponse(nodeId, ses.getId(),
+                             jr.getJobContext().getJobId(), null, null, null, null, null, null, false);
+ 
+                         fakeRes.setFakeException(new ClusterTopologyException("Node has left grid: " + nodeId));
+ 
+                         if (resList == null)
+                             resList = new ArrayList<>();
+ 
+                         resList.add(fakeRes);
+                     }
+                 }
+             }
+         }
+ 
+         if (resList == null)
+             return;
+ 
+         // Simulate responses without holding synchronization.
+         for (GridJobExecuteResponse res : resList) {
+             if (log.isDebugEnabled())
+                 log.debug("Simulating fake response from left node [res=" + res + ", nodeId=" + nodeId + ']');
+ 
+             onResponse(res);
+         }
+     }
+ 
+     /**
+      * @param evtType Event type.
+      * @param msg Event message.
+      */
+     private void recordTaskEvent(int evtType, String msg) {
+         if (!internal && ctx.event().isRecordable(evtType)) {
+             IgniteEvent evt = new IgniteTaskEvent(
+                 ctx.discovery().localNode(),
+                 msg,
+                 evtType,
+                 ses.getId(),
+                 ses.getTaskName(),
+                 ses.getTaskClassName(),
+                 internal,
+                 subjId);
+ 
+             ctx.event().record(evt);
+         }
+     }
+ 
+     /**
+      * @param evtType Event type.
+      * @param jobId Job ID.
+      * @param evtNode Event node.
+      * @param msg Event message.
+      */
+     private void recordJobEvent(int evtType, IgniteUuid jobId, ClusterNode evtNode, String msg) {
+         if (ctx.event().isRecordable(evtType)) {
+             IgniteJobEvent evt = new IgniteJobEvent();
+ 
+             evt.message(msg);
+             evt.node(ctx.discovery().localNode());
+             evt.taskName(ses.getTaskName());
+             evt.taskClassName(ses.getTaskClassName());
+             evt.taskSessionId(ses.getId());
+             evt.taskNode(evtNode);
+             evt.jobId(jobId);
+             evt.type(evtType);
+             evt.taskSubjectId(ses.subjectId());
+ 
+             ctx.event().record(evt);
+         }
+     }
+ 
+     /**
+      * @return Collection of job results.
+      */
+     private List<ComputeJobResult> getRemoteResults() {
+         assert Thread.holdsLock(mux);
+ 
+         List<ComputeJobResult> results = new ArrayList<>(jobRes.size());
+ 
+         for (GridJobResultImpl jobResult : jobRes.values())
+             if (jobResult.hasResponse())
+                 results.add(jobResult);
+ 
+         return results;
+     }
+ 
+     /**
+      * @param res Task result.
+      * @param e Exception.
+      */
+     void finishTask(@Nullable R res, @Nullable Throwable e) {
+         finishTask(res, e, true);
+     }
+ 
+     /**
+      * @param res Task result.
+      * @param e Exception.
+      * @param cancelChildren Whether to cancel children in case the task become cancelled.
+      */
+     void finishTask(@Nullable R res, @Nullable Throwable e, boolean cancelChildren) {
+         // Avoid finishing a job more than once from
+         // different threads.
+         synchronized (mux) {
+             if (state == State.REDUCING || state == State.FINISHING)
+                 return;
+ 
+             state = State.FINISHING;
+         }
+ 
+         try {
+             if (e == null)
+                 recordTaskEvent(EVT_TASK_FINISHED, "Task finished.");
+             else
+                 recordTaskEvent(EVT_TASK_FAILED, "Task failed.");
+ 
+             // Clean resources prior to finishing future.
+             evtLsnr.onTaskFinished(this);
+ 
+             if (cancelChildren)
+                 cancelChildren();
+         }
+         // Once we marked task as 'Finishing' we must complete it.
+         finally {
+             fut.onDone(res, e);
+ 
+             ses.onDone();
+         }
+     }
+ 
+     /**
+      * Checks whether node is alive or dead.
+      *
+      * @param uid UID of node to check.
+      * @return {@code true} if node is dead, {@code false} is node is alive.
+      */
+     private boolean isDeadNode(UUID uid) {
+         return ctx.discovery().node(uid) == null || !ctx.discovery().pingNode(uid);
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public boolean equals(Object obj) {
+         if (this == obj)
+             return true;
+ 
+         if (obj == null)
+             return false;
+ 
+         assert obj instanceof GridTaskWorker;
+ 
+         return ses.getId().equals(((GridTaskWorker<T, R>)obj).ses.getId());
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public int hashCode() {
+         return ses.getId().hashCode();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public String toString() {
 -        return S.toString(GridTaskWorker.class, this);
++        synchronized (mux) {
++            return S.toString(GridTaskWorker.class, this);
++        }
+     }
+ }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/shmem/GridIpcSharedMemoryNativeLoader.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/util/ipc/shmem/GridIpcSharedMemoryNativeLoader.java
index 0000000,2795e03..7bcb9e2
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/shmem/GridIpcSharedMemoryNativeLoader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/shmem/GridIpcSharedMemoryNativeLoader.java
@@@ -1,0 -1,243 +1,242 @@@
+ /*
+  * 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.util.ipc.shmem;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.internal.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ 
+ import java.io.*;
+ import java.net.*;
+ import java.nio.channels.*;
+ import java.util.*;
+ 
+ /**
+  * Shared memory native loader.
+  */
+ @SuppressWarnings("ErrorNotRethrown")
+ public class GridIpcSharedMemoryNativeLoader {
+     /** Loaded flag. */
+     private static volatile boolean loaded;
+ 
+     /** Library name base. */
+     private static final String LIB_NAME_BASE = "ggshmem";
+ 
+     /** Library name. */
+     private static final String LIB_NAME = LIB_NAME_BASE + "-" + GridProductImpl.VER;
+ 
+     /** Lock file path. */
+     private static final File LOCK_FILE = new File(System.getProperty("java.io.tmpdir"), "ggshmem.lock");
+ 
+     /**
+      * @return Operating system name to resolve path to library.
+      */
+     private static String os() {
+         String name = System.getProperty("os.name").toLowerCase().trim();
+ 
+         if (name.startsWith("win"))
+             throw new IllegalStateException("IPC shared memory native loader should not be called on windows.");
+ 
+         if (name.startsWith("linux"))
+             return "linux";
+ 
+         if (name.startsWith("mac os x"))
+             return "osx";
+ 
+         return name.replaceAll("\\W+", "_");
+     }
+ 
+     /**
+      * @return Platform.
+      */
+     private static String platform() {
+         return os() + bitModel();
+     }
+ 
+     /**
+      * @return Bit model.
+      */
+     private static int bitModel() {
+         String prop = System.getProperty("sun.arch.data.model");
+ 
+         if (prop == null)
+             prop = System.getProperty("com.ibm.vm.bitmode");
+ 
+         if (prop != null)
+             return Integer.parseInt(prop);
+ 
+         // We don't know.
+         return -1;
+     }
+ 
+     /**
+      * @throws IgniteCheckedException If failed.
+      */
+     public static void load() throws IgniteCheckedException {
+         if (loaded)
+             return;
+ 
+         synchronized (GridIpcSharedMemoryNativeLoader.class) {
+             if (loaded)
+                 return;
+ 
+             doLoad();
+ 
+             loaded = true;
+         }
+     }
+ 
+     /**
+      * @throws IgniteCheckedException If failed.
+      */
+     private static void doLoad() throws IgniteCheckedException {
+         assert Thread.holdsLock(GridIpcSharedMemoryNativeLoader.class);
+ 
+         Collection<Throwable> errs = new LinkedList<>();
+ 
+         try {
+             // Load native library (the library directory should be in java.library.path).
+             System.loadLibrary(LIB_NAME);
+ 
+             return;
+         }
+         catch (UnsatisfiedLinkError e) {
+             errs.add(e);
+         }
+ 
+         // Obtain lock on file to prevent concurrent extracts.
+         try (RandomAccessFile randomAccessFile = new RandomAccessFile(LOCK_FILE, "rws");
 -             FileLock lock = randomAccessFile.getChannel().lock()) {
 -
++             FileLock ignored = randomAccessFile.getChannel().lock()) {
+             if (extractAndLoad(errs, platformSpecificResourcePath()))
+                 return;
+ 
+             if (extractAndLoad(errs, osSpecificResourcePath()))
+                 return;
+ 
+             if (extractAndLoad(errs, resourcePath()))
+                 return;
+ 
+             // Failed to find the library.
+             assert !errs.isEmpty();
+ 
+             throw new IgniteCheckedException("Failed to load native IPC library: " + errs);
+         }
+         catch (IOException e) {
+             throw new IgniteCheckedException("Failed to obtain file lock: " + LOCK_FILE, e);
+         }
+     }
+ 
+     /**
+      * @return OS resource path.
+      */
+     private static String osSpecificResourcePath() {
+         return "META-INF/native/" + os() + "/" + mapLibraryName(LIB_NAME_BASE);
+     }
+ 
+     /**
+      * @return Platform resource path.
+      */
+     private static String platformSpecificResourcePath() {
+         return "META-INF/native/" + platform() + "/" + mapLibraryName(LIB_NAME_BASE);
+     }
+ 
+     /**
+      * @return Resource path.
+      */
+     private static String resourcePath() {
+         return "META-INF/native/" + mapLibraryName(LIB_NAME_BASE);
+     }
+ 
+     /**
+      * @return Maps library name to file name.
+      */
+     private static String mapLibraryName(String name) {
+         String libName = System.mapLibraryName(name);
+ 
+         if (U.isMacOs() && libName.endsWith(".jnilib"))
+             return libName.substring(0, libName.length() - "jnilib".length()) + "dylib";
+ 
+         return libName;
+     }
+ 
+     /**
+      * @param errs Errors collection.
+      * @param rsrcPath Path.
+      * @return {@code True} if library was found and loaded.
+      */
+     private static boolean extractAndLoad(Collection<Throwable> errs, String rsrcPath) {
+         ClassLoader clsLdr = U.detectClassLoader(GridIpcSharedMemoryNativeLoader.class);
+ 
+         URL rsrc = clsLdr.getResource(rsrcPath);
+ 
+         if (rsrc != null)
+             return extract(errs, rsrc, new File(System.getProperty("java.io.tmpdir"), mapLibraryName(LIB_NAME)));
+         else {
+             errs.add(new IllegalStateException("Failed to find resource with specified class loader " +
+                 "[rsrc=" + rsrcPath + ", clsLdr=" + clsLdr + ']'));
+ 
+             return false;
+         }
+     }
+ 
+     /**
+      * @param errs Errors collection.
+      * @param src Source.
+      * @param target Target.
+      * @return {@code True} if resource was found and loaded.
+      */
+     @SuppressWarnings("ResultOfMethodCallIgnored")
+     private static boolean extract(Collection<Throwable> errs, URL src, File target) {
+         FileOutputStream os = null;
+         InputStream is = null;
+ 
+         try {
+             if (!target.exists()) {
+                 is = src.openStream();
+ 
+                 if (is != null) {
+                     os = new FileOutputStream(target);
+ 
+                     int read;
+ 
+                     byte[] buf = new byte[4096];
+ 
+                     while ((read = is.read(buf)) != -1)
+                         os.write(buf, 0, read);
+                 }
+             }
+ 
+             // chmod 775.
+             if (!U.isWindows())
+                 Runtime.getRuntime().exec(new String[] {"chmod", "775", target.getCanonicalPath()}).waitFor();
+ 
+             System.load(target.getPath());
+ 
+             return true;
+         }
+         catch (IOException | UnsatisfiedLinkError | InterruptedException e) {
+             errs.add(e);
+         }
+         finally {
+             U.closeQuiet(os);
+             U.closeQuiet(is);
+         }
+ 
+         return false;
+     }
+ }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerSelfTest.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerSelfTest.java
index 0000000,55fa051..538682e
mode 000000,100644..100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/GridIoManagerSelfTest.java
@@@ -1,0 -1,249 +1,249 @@@
+ /*
+  * 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.communication;
+ 
+ import org.apache.commons.collections.*;
+ import org.apache.ignite.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.internal.*;
+ import org.apache.ignite.marshaller.jdk.*;
+ import org.apache.ignite.spi.communication.tcp.*;
+ import org.apache.ignite.internal.managers.discovery.*;
+ import org.apache.ignite.internal.util.direct.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.testframework.*;
+ import org.apache.ignite.testframework.junits.*;
+ import org.apache.ignite.testframework.junits.common.*;
+ import org.mockito.*;
+ 
+ import java.io.*;
+ import java.nio.*;
+ import java.util.*;
+ import java.util.concurrent.*;
+ 
+ import static org.mockito.Matchers.any;
+ import static org.mockito.Mockito.anyLong;
+ import static org.mockito.Mockito.argThat;
+ import static org.mockito.Mockito.eq;
+ import static org.mockito.Mockito.*;
+ 
+ /**
+  * Test for {@link GridIoManager}.
+  */
+ public class GridIoManagerSelfTest extends GridCommonAbstractTest {
+     /** Grid test context. */
+     private GridTestKernalContext ctx = new GridTestKernalContext(log);
+ 
+     /** Test local node. */
+     private GridTestNode locNode = new GridTestNode(UUID.randomUUID());
+ 
+     /** Test remote node. */
+     private GridTestNode rmtNode = new GridTestNode(UUID.randomUUID());
+ 
+     /** {@inheritDoc} */
+     @Override protected void beforeTest() throws Exception {
+         ctx.config().setCommunicationSpi(new TcpCommunicationSpi());
+         ctx.config().setMarshaller(new IgniteJdkMarshaller());
+ 
+         // Turn off peer class loading to simplify mocking.
+         ctx.config().setPeerClassLoadingEnabled(false);
+ 
+         // Register local and remote nodes in discovery manager.
+         GridDiscoveryManager mockedDiscoveryMgr = Mockito.mock(GridDiscoveryManager.class);
+ 
+         when(mockedDiscoveryMgr.localNode()).thenReturn(locNode);
+         when(mockedDiscoveryMgr.remoteNodes()).thenReturn(F.<ClusterNode>asList(rmtNode));
+ 
+         ctx.add(mockedDiscoveryMgr);
+     }
+ 
+     /**
+      * @throws Exception If failed.
+      */
+     public void testSendIfOneOfNodesIsLocalAndTopicIsEnum() throws Exception {
+         GridTestUtils.assertThrows(log, new Callable<Object>() {
+             @Override public Object call() throws Exception {
+                 new GridIoManager(ctx).send(F.asList(locNode, rmtNode), GridTopic.TOPIC_CACHE, new Message(),
+                     GridIoPolicy.P2P_POOL);
+ 
+                 return null;
+             }
+         }, AssertionError.class, "Internal GridGain code should never call the method with local node in a node list.");
+     }
+ 
+     /**
+      * @throws Exception If failed.
+      */
+     public void testSendIfOneOfNodesIsLocalAndTopicIsObject() throws Exception {
+         GridTestUtils.assertThrows(log, new Callable<Object>() {
+             @Override public Object call() throws Exception {
+                 new GridIoManager(ctx).send(F.asList(locNode, rmtNode), new Object(), new Message(),
+                     GridIoPolicy.P2P_POOL);
+ 
+                 return null;
+             }
+         }, AssertionError.class, "Internal GridGain code should never call the method with local node in a node list.");
+     }
+ 
+     /**
+      * @throws Exception If failed.
+      */
+     public void testSendUserMessageThinVersionIfOneOfNodesIsLocal() throws Exception {
+         Object msg = new Object();
+ 
+         GridIoManager ioMgr = spy(new TestGridIoManager(ctx));
+ 
+         try {
+             ioMgr.sendUserMessage(F.asList(locNode, rmtNode), msg);
+         }
+         catch (IgniteCheckedException ignored) {
+             // No-op. We are using mocks so real sending is impossible.
+         }
+ 
+         verify(ioMgr).send(eq(locNode), eq(GridTopic.TOPIC_COMM_USER), any(GridIoUserMessage.class),
+             eq(GridIoPolicy.PUBLIC_POOL));
+ 
+         Collection<? extends ClusterNode> rmtNodes = F.view(F.asList(rmtNode), F.remoteNodes(locNode.id()));
+ 
+         verify(ioMgr).send(argThat(new IsEqualCollection(rmtNodes)), eq(GridTopic.TOPIC_COMM_USER),
+             any(GridIoUserMessage.class), eq(GridIoPolicy.PUBLIC_POOL));
+     }
+ 
+     /**
+      * @throws Exception If failed.
+      */
+     public void testSendUserMessageUnorderedThickVersionIfOneOfNodesIsLocal() throws Exception {
+         Object msg = new Object();
+ 
+         GridIoManager ioMgr = spy(new TestGridIoManager(ctx));
+ 
+         try {
+             ioMgr.sendUserMessage(F.asList(locNode, rmtNode), msg, GridTopic.TOPIC_GGFS, false, 123L);
+         }
+         catch (IgniteCheckedException ignored) {
+             // No-op. We are using mocks so real sending is impossible.
+         }
+ 
+         verify(ioMgr).send(eq(locNode), eq(GridTopic.TOPIC_COMM_USER), any(GridIoUserMessage.class),
+             eq(GridIoPolicy.PUBLIC_POOL));
+ 
+         Collection<? extends ClusterNode> rmtNodes = F.view(F.asList(rmtNode), F.remoteNodes(locNode.id()));
+ 
+         verify(ioMgr).send(argThat(new IsEqualCollection(rmtNodes)), eq(GridTopic.TOPIC_COMM_USER),
+             any(GridIoUserMessage.class), eq(GridIoPolicy.PUBLIC_POOL));
+     }
+ 
+     /**
+      * @throws Exception If failed.
+      */
+     public void testSendUserMessageOrderedThickVersionIfOneOfNodesIsLocal() throws Exception {
+         Object msg = new Object();
+ 
+         GridIoManager ioMgr = spy(new TestGridIoManager(ctx));
+ 
+         try {
+             ioMgr.sendUserMessage(F.asList(locNode, rmtNode), msg, GridTopic.TOPIC_GGFS, true, 123L);
+         }
+         catch (Exception ignored) {
+             // No-op. We are using mocks so real sending is impossible.
+         }
+ 
+         verify(ioMgr).sendOrderedMessage(
+             argThat(new IsEqualCollection(F.asList(locNode, rmtNode))),
 -            eq(GridTopic.TOPIC_COMM_USER), anyLong(),
++            eq(GridTopic.TOPIC_COMM_USER),
+             any(GridIoUserMessage.class),
+             eq(GridIoPolicy.PUBLIC_POOL),
+             eq(123L),
+             false);
+     }
+ 
+     /**
+      * Test-purposed extension of {@code GridIoManager} with no-op {@code send(...)} methods.
+      */
+     private static class TestGridIoManager extends GridIoManager {
+         /**
+          * @param ctx Grid kernal context.
+          */
+         TestGridIoManager(GridKernalContext ctx) {
+             super(ctx);
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public void send(ClusterNode node, GridTopic topic, GridTcpCommunicationMessageAdapter msg,
+             GridIoPolicy plc) throws IgniteCheckedException {
+             // No-op.
+         }
+     }
+ 
+     /**
+      * Mockito argument matcher to compare collections produced by {@code F.view()} methods.
+      */
+     private static class IsEqualCollection extends ArgumentMatcher<Collection<? extends ClusterNode>> {
+         /** Expected collection. */
+         private final Collection<? extends ClusterNode> expCol;
+ 
+         /**
+          * Default constructor.
+          *
+          * @param expCol Expected collection.
+          */
+         IsEqualCollection(Collection<? extends ClusterNode> expCol) {
+             this.expCol = expCol;
+         }
+ 
+         /**
+          * Matches a given collection to the specified in constructor expected one
+          * with Apache {@code CollectionUtils.isEqualCollection()}.
+          *
+          * @param colToCheck Collection to be matched against the expected one.
+          * @return True if collections matches.
+          */
+         @Override public boolean matches(Object colToCheck) {
+             return CollectionUtils.isEqualCollection(expCol, (Collection)colToCheck);
+         }
+     }
+ 
+     /** */
+     private static class Message extends GridTcpCommunicationMessageAdapter implements Serializable {
+         /** {@inheritDoc} */
+         @SuppressWarnings("CloneDoesntCallSuperClone")
+         @Override public GridTcpCommunicationMessageAdapter clone() {
+             throw new UnsupportedOperationException();
+         }
+ 
+         /** {@inheritDoc} */
+         @Override protected void clone0(GridTcpCommunicationMessageAdapter _msg) {
+             // No-op.
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public boolean writeTo(ByteBuffer buf) {
+             return true;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public boolean readFrom(ByteBuffer buf) {
+             return true;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public byte directType() {
+             return 0;
+         }
+     }
+ }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheSyncReplicatedPreloadSelfTest.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheSyncReplicatedPreloadSelfTest.java
index 0000000,7cd698d..a30ef73
mode 000000,100644..100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheSyncReplicatedPreloadSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheSyncReplicatedPreloadSelfTest.java
@@@ -1,0 -1,166 +1,165 @@@
+ /*
+  * 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.cache.distributed.replicated;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cache.*;
+ import org.apache.ignite.configuration.*;
+ import org.apache.ignite.spi.discovery.tcp.*;
+ import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+ import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+ import org.apache.ignite.testframework.junits.common.*;
+ import org.jetbrains.annotations.*;
+ 
+ import java.util.concurrent.*;
+ import java.util.concurrent.atomic.*;
+ 
+ import static org.apache.ignite.configuration.IgniteDeploymentMode.*;
+ import static org.apache.ignite.cache.CacheDistributionMode.*;
+ import static org.apache.ignite.cache.CacheMode.*;
+ import static org.apache.ignite.cache.CachePreloadMode.*;
+ import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
+ 
+ /**
+  * Multithreaded tests for replicated cache preloader.
+  */
+ public class GridCacheSyncReplicatedPreloadSelfTest extends GridCommonAbstractTest {
+     /** */
+     private TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+ 
+     /** */
+     private static final boolean DISCO_DEBUG_MODE = false;
+ 
 -
+     /**
+      * Constructs test.
+      */
+     public GridCacheSyncReplicatedPreloadSelfTest() {
+         super(false /* don't start grid. */);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+         IgniteConfiguration cfg = super.getConfiguration(gridName);
+ 
+         TcpDiscoverySpi disco = new TcpDiscoverySpi();
+ 
+         disco.setIpFinder(ipFinder);
+         disco.setDebugMode(DISCO_DEBUG_MODE);
+ 
+         cfg.setDiscoverySpi(disco);
+ 
+         CacheConfiguration cacheCfg = defaultCacheConfiguration();
+ 
+         cacheCfg.setCacheMode(REPLICATED);
+         cacheCfg.setDistributionMode(PARTITIONED_ONLY);
+         cacheCfg.setWriteSynchronizationMode(FULL_SYNC);
+ 
+         // This property is essential for this test.
+         cacheCfg.setPreloadMode(SYNC);
+ 
+         cacheCfg.setPreloadBatchSize(10000);
+ 
+         cfg.setCacheConfiguration(cacheCfg);
+         cfg.setDeploymentMode(CONTINUOUS);
+ 
+         return cfg;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected void afterTest() throws Exception {
+         super.afterTest();
+ 
+         stopAllGrids();
+     }
+ 
+     /**
+      * @throws Exception If test failed.
+      */
+     @SuppressWarnings({"TooBroadScope"})
+     public void testNodeRestart() throws Exception {
+         int keyCnt = 1000;
+         int retries = 20;
+ 
+         Ignite g0 = startGrid(0);
+         Ignite g1 = startGrid(1);
+ 
+         for (int i = 0; i < keyCnt; i++)
+             g0.cache(null).putx(i, i);
+ 
+         assertEquals(keyCnt, g0.cache(null).size());
+         assertEquals(keyCnt, g1.cache(null).size());
+ 
+         for (int n = 0; n < retries; n++) {
+             info("Starting additional grid node...");
+ 
+             Ignite g2 = startGrid(2);
+ 
+             assertEquals(keyCnt, g2.cache(null).size());
+ 
+             info("Stopping additional grid node...");
+ 
+             stopGrid(2);
+         }
+     }
+ 
+     /**
+      * @throws Exception If test failed.
+      */
+     @SuppressWarnings({"TooBroadScope"})
+     public void testNodeRestartMultithreaded() throws Exception {
+         final int keyCnt = 1000;
+         final int retries = 300;
+         int threadCnt = 5;
+ 
+         Ignite g0 = startGrid(0);
+         Ignite g1 = startGrid(1);
+ 
+         for (int i = 0; i < keyCnt; i++)
+             g0.cache(null).putx(i, i);
+ 
+         assertEquals(keyCnt, g0.cache(null).size());
+         assertEquals(keyCnt, g1.cache(null).size());
+ 
+         final AtomicInteger cnt = new AtomicInteger();
+ 
+         multithreaded(
+             new Callable() {
+                 @Nullable @Override public Object call() throws Exception {
+                     while (true) {
+                         int c = cnt.incrementAndGet();
+ 
+                         if (c > retries)
+                             break;
+ 
+                         int idx = c + 1;
+ 
+                         info("Starting additional grid node with index: " + idx);
+ 
+                         startGrid(idx);
+ 
+                         info("Stopping additional grid node with index: " + idx);
+ 
+                         stopGrid(idx);
+                     }
+ 
+                     return null;
+                 }
+             },
+             threadCnt);
+     }
+ }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/test/java/org/apache/ignite/session/GridSessionLoadSelfTest.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/session/GridSessionLoadSelfTest.java
index 0000000,e4de992..d148d37
mode 000000,100644..100644
--- a/modules/core/src/test/java/org/apache/ignite/session/GridSessionLoadSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/session/GridSessionLoadSelfTest.java
@@@ -1,0 -1,264 +1,267 @@@
+ /*
+  * 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.session;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.compute.*;
+ import org.apache.ignite.configuration.*;
+ import org.apache.ignite.resources.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ import org.apache.ignite.testframework.*;
+ import org.apache.ignite.testframework.junits.common.*;
+ 
+ import java.io.*;
+ import java.util.*;
+ import java.util.concurrent.*;
+ 
+ /**
+  * Task session load self test.
+  */
+ @GridCommonTest(group = "Task Session")
+ public class GridSessionLoadSelfTest extends GridCommonAbstractTest {
+     /** */
+     private static final int THREAD_CNT = 40;
+ 
+     /** */
+     private static final int EXEC_CNT = 10;
+ 
+     /** */
+     private boolean locMarsh;
+ 
+     /** {@inheritDoc} */
+     @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+         IgniteConfiguration c = super.getConfiguration(gridName);
+ 
+         c.setMarshalLocalJobs(locMarsh);
+         c.setPeerClassLoadingEnabled(false);
+ 
+         return c;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected void beforeTest() throws Exception {
+         startGrids(2);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected void afterTest() throws Exception {
+         stopAllGrids();
+     }
+ 
+     /**
+      * @throws Exception If failed.
+      */
+     public void testSessionLoad() throws Exception {
+         locMarsh = true;
+ 
+         checkSessionLoad();
+     }
+ 
+     /**
+      * @throws Exception If failed.
+      */
+     public void testSessionLoadNoLocalMarshalling() throws Exception {
+         locMarsh = false;
+ 
+         checkSessionLoad();
+     }
+ 
+     /**
+      * @throws Exception If failed.
+      */
+     private void checkSessionLoad() throws Exception {
 -        final Ignite ignite = grid(1);
++        final Ignite ignite = grid(0);
+ 
+         assert ignite != null;
+         assert ignite.cluster().nodes().size() == 2;
+ 
+         info("Thread count: " + THREAD_CNT);
+ 
+         GridTestUtils.runMultiThreaded(new Callable<Object>() {
+             @Override public Object call() throws Exception {
++                ComputeTaskFuture f = null;
++
+                 try {
+                     for (int i = 0; i < EXEC_CNT; i++)
+                         assertEquals(Boolean.TRUE,
 -                            executeAsync(ignite.compute().withName("task-name"),
++                            (f = executeAsync(ignite.compute().withName("task-name"),
+                                 SessionLoadTestTask.class,
 -                                ignite.cluster().nodes().size() * 2).get(20000));
++                                ignite.cluster().nodes().size() * 2)).get(20000));
+                 }
+                 catch (Exception e) {
 -                    U.error(log, "Test failed.", e);
++                    U.error(log, "Task failed: " +
++                        f != null ? f.getTaskSession().getId() : "N/A", e);
+ 
+                     throw e;
+                 }
+                 finally {
+                     info("Thread finished.");
+                 }
+ 
+                 return null;
+             }
+         }, THREAD_CNT, "grid-load-test-thread");
+     }
+ 
+     /**
+      *
+      */
+     @ComputeTaskSessionFullSupport
+     private static class SessionLoadTestTask extends ComputeTaskAdapter<Integer, Boolean> {
+         /** */
+         @IgniteTaskSessionResource
+         private ComputeTaskSession taskSes;
+ 
+         /** */
+         @IgniteLoggerResource
+         private IgniteLogger log;
+ 
+         /** */
+         private Map<String, Integer> params;
+ 
+         /** {@inheritDoc} */
+         @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, Integer arg)
+             throws IgniteCheckedException {
+             assert taskSes != null;
+             assert arg != null;
+             assert arg > 1;
+ 
+             Map<SessionLoadTestJob, ClusterNode> map = new HashMap<>(subgrid.size());
+ 
+             Iterator<ClusterNode> iter = subgrid.iterator();
+ 
+             Random rnd = new Random();
+ 
+             params = new HashMap<>(arg);
+ 
+             for (int i = 0; i < arg; i++) {
+                 // Recycle iterator.
+                 if (!iter.hasNext())
+                     iter = subgrid.iterator();
+ 
+                 String paramName = UUID.randomUUID().toString();
+ 
+                 int paramVal = rnd.nextInt();
+ 
+                 taskSes.setAttribute(paramName, paramVal);
+ 
+                 map.put(new SessionLoadTestJob(paramName), iter.next());
+ 
+                 params.put(paramName, paramVal);
+ 
+                 if (log.isDebugEnabled())
+                     log.debug("Set session attribute [name=" + paramName + ", value=" + paramVal + ']');
+             }
+ 
+             return map;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public Boolean reduce(List<ComputeJobResult> results) throws IgniteCheckedException {
+             assert taskSes != null;
+             assert results != null;
+             assert params != null;
+             assert !params.isEmpty();
+             assert results.size() == params.size();
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Reducing: " + params);
+ 
+             Map<String, Integer> receivedParams = new HashMap<>();
+ 
+             boolean allAttrReceived = false;
+ 
+             for (int i = 0; i < 3 && !allAttrReceived; i++) {
+                 allAttrReceived = true;
+ 
+                 for (Map.Entry<String, Integer> entry : params.entrySet()) {
+                     Serializable attr = taskSes.getAttribute(entry.getKey());
+ 
+                     assert attr != null;
+ 
+                     int newVal = (Integer)attr;
+ 
+                     receivedParams.put(entry.getKey(), newVal);
+ 
+                     // New value is expected to be +1 to argument value.
+                     if (newVal != entry.getValue() + 1)
+                         allAttrReceived = false;
+                 }
+ 
+                 if (!allAttrReceived)
+                     U.sleep(1000);
+             }
+ 
+             if (log.isDebugEnabled()) {
+                 for (Map.Entry<String, Integer> entry : receivedParams.entrySet())
+                     log.debug("Received session attribute value [name=" + entry.getKey() + ", val=" + entry.getValue()
+                         + ", expected=" + (params.get(entry.getKey()) + 1) + ']');
+             }
+ 
+             return allAttrReceived;
+         }
+     }
+ 
+     /**
+      *
+      */
+     private static class SessionLoadTestJob extends ComputeJobAdapter {
+         /** */
+         @IgniteTaskSessionResource
+         private ComputeTaskSession taskSes;
+ 
+         /** */
+         @IgniteLoggerResource
+         private IgniteLogger log;
+ 
+         /**
+          * @param arg Argument.
+          */
+         private SessionLoadTestJob(String arg) {
+             super(arg);
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public Serializable execute() throws IgniteCheckedException {
+             assert taskSes != null;
+             assert argument(0) != null;
+ 
+             Serializable ser = taskSes.getAttribute(argument(0));
+ 
+             assert ser != null;
+ 
+             int val = (Integer)ser + 1;
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Executing session load job: " + val);
+ 
+             // Generate garbage.
+             for (int i = 0; i < 10; i++)
+                 taskSes.setAttribute(argument(0), i);
+ 
+             // Set final value (+1 to original value).
+             taskSes.setAttribute(argument(0), val);
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Set session attribute [name=" + argument(0) + ", value=" + val + ']');
+ 
+             return val;
+         }
+     }
+ }


Mime
View raw message