http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
----------------------------------------------------------------------
diff --cc
modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
index 0000000,477e742..93026e8
mode 000000,100644..100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
@@@ -1,0 -1,1868 +1,1859 @@@
+ /*
+ * 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.job;
+
+ import org.apache.ignite.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.compute.*;
+ import org.apache.ignite.events.*;
+ import org.apache.ignite.internal.*;
+ import org.apache.ignite.internal.processors.*;
+ import org.apache.ignite.internal.util.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.marshaller.*;
+ import org.apache.ignite.internal.managers.collision.*;
+ import org.apache.ignite.internal.managers.communication.*;
+ import org.apache.ignite.internal.managers.deployment.*;
+ import org.apache.ignite.internal.managers.eventstorage.*;
+ import org.apache.ignite.internal.processors.jobmetrics.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ import org.jdk8.backport.*;
+ import org.jetbrains.annotations.*;
+
+ import java.io.*;
+ import java.util.*;
+ import java.util.concurrent.*;
+ import java.util.concurrent.atomic.*;
+ import java.util.concurrent.locks.*;
+
+ import static java.util.concurrent.TimeUnit.*;
+ import static org.apache.ignite.IgniteSystemProperties.*;
+ 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.jdk8.backport.ConcurrentLinkedHashMap.QueuePolicy.*;
+
+ /**
+ * Responsible for all grid job execution and communication.
+ */
+ public class GridJobProcessor extends GridProcessorAdapter {
+ /** */
+ private static final int FINISHED_JOBS_COUNT =
Integer.getInteger(GG_JOBS_HISTORY_SIZE, 10240);
+
+ /** */
+ private final IgniteMarshaller marsh;
+
+ /** */
+ private final boolean jobAlwaysActivate;
+
+ /** */
+ private final ConcurrentMap<IgniteUuid, GridJobWorker> activeJobs;
+
+ /** */
+ private final ConcurrentMap<IgniteUuid, GridJobWorker> passiveJobs;
+
+ /** */
+ private final ConcurrentMap<IgniteUuid, GridJobWorker> cancelledJobs =
+ new ConcurrentHashMap8<>();
+
+ /** */
+ private final Collection<IgniteUuid> heldJobs = new
GridConcurrentHashSet<>();
+
+ /** If value is {@code true}, job was cancelled from future. */
+ private final GridBoundedConcurrentLinkedHashMap<IgniteUuid, Boolean>
cancelReqs =
+ new GridBoundedConcurrentLinkedHashMap<>(FINISHED_JOBS_COUNT,
+ FINISHED_JOBS_COUNT < 128 ? FINISHED_JOBS_COUNT : 128,
+ 0.75f, 16);
+
+ /** */
+ private final GridBoundedConcurrentLinkedHashSet<IgniteUuid> finishedJobs
=
+ new GridBoundedConcurrentLinkedHashSet<>(FINISHED_JOBS_COUNT,
+ FINISHED_JOBS_COUNT < 128 ? FINISHED_JOBS_COUNT : 128,
+ 0.75f, 256, PER_SEGMENT_Q);
+
+ /** */
+ private final GridJobEventListener evtLsnr;
+
+ /** */
+ private final GridMessageListener cancelLsnr;
+
+ /** */
+ private final GridMessageListener jobExecLsnr;
+
+ /** */
+ private final GridLocalEventListener discoLsnr;
+
+ /** Needed for statistics. */
+ private final LongAdder canceledJobsCnt = new LongAdder();
+
+ /** Needed for statistics. */
+ private final LongAdder finishedJobsCnt = new LongAdder();
+
+ /** Needed for statistics. */
+ private final LongAdder startedJobsCnt = new LongAdder();
+
+ /** Needed for statistics. */
+ private final LongAdder rejectedJobsCnt = new LongAdder();
+
+ /** Total job execution time (unaccounted for in metrics). */
+ private final LongAdder finishedJobsTime = new LongAdder();
+
+ /** Maximum job execution time for finished jobs. */
+ private final GridAtomicLong maxFinishedJobsTime = new GridAtomicLong();
+
+ /** */
+ private final AtomicLong metricsLastUpdateTstamp = new AtomicLong();
+
+ /** */
+ private boolean stopping;
+
+ /** */
+ private boolean cancelOnStop;
+
+ /** */
+ private final long metricsUpdateFreq;
+
+ /** */
+ private final GridSpinReadWriteLock rwLock = new GridSpinReadWriteLock();
+
+ /** Topic ID generator. */
+ private final AtomicLong topicIdGen = new AtomicLong();
+
+ /** */
+ private final GridJobHoldListener holdLsnr = new JobHoldListener();
+
+ /** */
+ private final ThreadLocal<Boolean> handlingCollision = new
ThreadLocal<Boolean>() {
+ @Override protected Boolean initialValue() {
+ return false;
+ }
+ };
+
+ /** Internal task flag. */
+ private final GridThreadLocal<Boolean> internal = new
GridThreadLocal<Boolean>() {
+ @Override protected Boolean initialValue() {
+ return false;
+ }
+ };
+
+ /** Current session. */
+ private final GridThreadLocal<ComputeTaskSession> currentSess = new
GridThreadLocal<>();
+
+ /**
+ * @param ctx Kernal context.
+ */
+ public GridJobProcessor(GridKernalContext ctx) {
+ super(ctx);
+
+ marsh = ctx.config().getMarshaller();
+
+ // Collision manager is already started and is fully functional.
+ jobAlwaysActivate = !ctx.collision().enabled();
+
+ metricsUpdateFreq = ctx.config().getMetricsUpdateFrequency();
+
+ activeJobs = jobAlwaysActivate ? new ConcurrentHashMap8<IgniteUuid,
GridJobWorker>() :
+ new JobsMap(1024, 0.75f, 256);
+
+ passiveJobs = jobAlwaysActivate ? null : new JobsMap(1024, 0.75f,
256);
+
+ evtLsnr = new JobEventListener();
+ cancelLsnr = new JobCancelListener();
+ jobExecLsnr = new JobExecutionListener();
+ discoLsnr = new JobDiscoveryListener();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void start() throws IgniteCheckedException {
+ if (metricsUpdateFreq < -1)
+ throw new IgniteCheckedException("Invalid value for
'metricsUpdateFrequency' configuration property " +
+ "(should be greater than or equals to -1): " +
metricsUpdateFreq);
+
+ if (metricsUpdateFreq == -1)
+ U.warn(log, "Job metrics are disabled (use with caution).");
+
+ if (!jobAlwaysActivate)
+ ctx.collision().setCollisionExternalListener(new
CollisionExternalListener());
+
+ GridIoManager ioMgr = ctx.io();
+
+ ioMgr.addMessageListener(TOPIC_JOB_CANCEL, cancelLsnr);
+ ioMgr.addMessageListener(TOPIC_JOB, jobExecLsnr);
+
+ ctx.event().addLocalEventListener(discoLsnr, EVT_NODE_FAILED,
EVT_NODE_LEFT, EVT_NODE_METRICS_UPDATED);
+
+ if (log.isDebugEnabled())
+ log.debug("Job processor started.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop(boolean cancel) {
+ // Clear collections.
+ activeJobs.clear();
+ cancelledJobs.clear();
+ cancelReqs.clear();
+
+ if (log.isDebugEnabled())
+ log.debug("Job processor stopped.");
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onKernalStop(boolean cancel) {
+ // Stop receiving new requests and sending responses.
+ GridIoManager commMgr = ctx.io();
+
+ commMgr.removeMessageListener(TOPIC_JOB, jobExecLsnr);
+ commMgr.removeMessageListener(TOPIC_JOB_CANCEL, cancelLsnr);
+
+ if (!jobAlwaysActivate)
+ // Ignore external collision events.
+ ctx.collision().unsetCollisionExternalListener();
+
+ rwLock.writeLock();
+
+ try {
+ stopping = true;
+
+ cancelOnStop = cancel;
+ }
+ finally {
+ rwLock.writeUnlock();
+ }
+
+ // Rejected jobs.
+ if (!jobAlwaysActivate) {
+ for (GridJobWorker job : passiveJobs.values())
+ if (passiveJobs.remove(job.getJobId(), job))
+ rejectJob(job, false);
+ }
+
+ // Cancel only if we force grid to stop
+ if (cancel) {
+ for (GridJobWorker job : activeJobs.values()) {
+ job.onStopping();
+
+ cancelJob(job, false);
+ }
+ }
+
+ U.join(activeJobs.values(), log);
+ U.join(cancelledJobs.values(), log);
+
+ // Ignore topology changes.
+ ctx.event().removeLocalEventListener(discoLsnr);
+
+ if (log.isDebugEnabled())
+ log.debug("Finished executing job processor onKernalStop()
callback.");
+ }
+
+ /**
+ * Gets active job.
+ *
+ * @param jobId Job ID.
+ * @return Active job.
+ */
+ @Nullable public GridJobWorker activeJob(IgniteUuid jobId) {
+ assert jobId != null;
+
+ return activeJobs.get(jobId);
+ }
+
+ /**
+ * @return {@code True} if running internal task.
+ */
+ public boolean internal() {
+ return internal.get();
+ }
+
+ /**
+ * Sets internal task flag.
+ *
+ * @param internal {@code True} if running internal task.
+ */
+ void internal(boolean internal) {
+ this.internal.set(internal);
+ }
+
+ /**
+ * @param job Rejected job.
+ * @param sndReply {@code True} to send reply.
+ */
+ private void rejectJob(GridJobWorker job, boolean sndReply) {
+ IgniteCheckedException e = new ComputeExecutionRejectedException("Job
was cancelled before execution [taskSesId=" +
+ job.getSession().getId() + ", jobId=" + job.getJobId() + ", job="
+ job.getJob() + ']');
+
+ job.finishJob(null, e, sndReply);
+ }
+
+ /**
+ * @param job Canceled job.
+ * @param sysCancel {@code True} if job has been cancelled from system
and no response needed.
+ */
+ private void cancelJob(GridJobWorker job, boolean sysCancel) {
+ boolean isCancelled = job.isCancelled();
+
+ // We don't increment number of cancelled jobs if it
+ // was already cancelled.
+ if (!job.isInternal() && !isCancelled)
+ canceledJobsCnt.increment();
+
+ job.cancel(sysCancel);
+ }
+
+ /**
+ * @param dep Deployment to release.
+ */
+ private void release(GridDeployment dep) {
+ dep.release();
+
+ if (dep.obsolete())
+ ctx.resource().onUndeployed(dep);
+ }
+
+ /**
+ * @param ses Session.
+ * @param attrs Attributes.
+ * @throws IgniteCheckedException If failed.
+ */
+ public void setAttributes(GridJobSessionImpl ses, Map<?, ?> attrs) throws
IgniteCheckedException {
+ assert ses.isFullSupport();
+
+ long timeout = ses.getEndTime() - U.currentTimeMillis();
+
+ if (timeout <= 0) {
+ U.warn(log, "Task execution timed out (remote session attributes
won't be set): " + ses);
+
+ return;
+ }
+
+ if (log.isDebugEnabled())
+ log.debug("Setting session attribute(s) from job: " + ses);
+
+ ClusterNode taskNode = ctx.discovery().node(ses.getTaskNodeId());
+
+ if (taskNode == null)
+ throw new IgniteCheckedException("Node that originated task
execution has left grid: " +
+ ses.getTaskNodeId());
+
+ boolean loc = ctx.localNodeId().equals(taskNode.id()) &&
!ctx.config().isMarshalLocalJobs();
+
+ GridTaskSessionRequest req = new GridTaskSessionRequest(ses.getId(),
ses.getJobId(),
+ loc ? null : marsh.marshal(attrs), attrs);
+
+ Object topic = TOPIC_TASK.topic(ses.getJobId(),
ctx.discovery().localNode().id());
+
+ // Always go through communication to preserve order.
+ ctx.io().sendOrderedMessage(
+ taskNode,
+ topic, // Job topic.
- ctx.io().nextMessageId(topic, taskNode.id()),
+ req,
+ SYSTEM_POOL,
+ timeout,
+ false);
+ }
+
+ /**
+ * @param ses Session.
+ * @return Siblings.
+ * @throws IgniteCheckedException If failed.
+ */
+ public Collection<ComputeJobSibling> requestJobSiblings(final
ComputeTaskSession ses) throws IgniteCheckedException {
+ assert ses != null;
+
+ final UUID taskNodeId = ses.getTaskNodeId();
+
+ ClusterNode taskNode = ctx.discovery().node(taskNodeId);
+
+ if (taskNode == null)
+ throw new IgniteCheckedException("Node that originated task
execution has left grid: " + taskNodeId);
+
+ // Tuple: error message-response.
+ final IgniteBiTuple<String, GridJobSiblingsResponse> t = F.t2();
+
+ final Lock lock = new ReentrantLock();
+ final Condition cond = lock.newCondition();
+
+ GridMessageListener msgLsnr = new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ String err = null;
+ GridJobSiblingsResponse res = null;
+
+ if (!(msg instanceof GridJobSiblingsResponse))
+ err = "Received unexpected message: " + msg;
+ else if (!nodeId.equals(taskNodeId))
+ err = "Received job siblings response from unexpected
node [taskNodeId=" + taskNodeId +
+ ", nodeId=" + nodeId + ']';
+ else
+ // Sender and message type are fine.
+ res = (GridJobSiblingsResponse)msg;
+
+ if (res.jobSiblings() == null) {
+ try {
+ res.unmarshalSiblings(marsh);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to unmarshal job siblings.", e);
+
+ err = e.getMessage();
+ }
+ }
+
+ lock.lock();
+
+ try {
+ if (t.isEmpty()) {
+ t.set(err, res);
+
+ cond.signalAll();
+ }
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+ };
+
+ GridLocalEventListener discoLsnr = new GridLocalEventListener() {
+ @Override public void onEvent(IgniteEvent evt) {
+ assert evt instanceof IgniteDiscoveryEvent &&
+ (evt.type() == EVT_NODE_FAILED || evt.type() ==
EVT_NODE_LEFT) : "Unexpected event: " + evt;
+
+ IgniteDiscoveryEvent discoEvt = (IgniteDiscoveryEvent)evt;
+
+ if (taskNodeId.equals(discoEvt.eventNode().id())) {
+ lock.lock();
+
+ try {
+ if (t.isEmpty()) {
+ t.set("Node that originated task execution has
left grid: " + taskNodeId, null);
+
+ cond.signalAll();
+ }
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+ }
+ };
+
+ boolean loc = ctx.localNodeId().equals(taskNodeId);
+
+ // 1. Create unique topic name.
+ Object topic = TOPIC_JOB_SIBLINGS.topic(ses.getId(),
topicIdGen.getAndIncrement());
+
+ try {
+ // 2. Register listener.
+ ctx.io().addMessageListener(topic, msgLsnr);
+
+ // 3. Send message.
+ ctx.io().send(taskNode, TOPIC_JOB_SIBLINGS,
+ new GridJobSiblingsRequest(ses.getId(),
+ loc ? topic : null,
+ loc ? null : marsh.marshal(topic)),
+ SYSTEM_POOL);
+
+ // 4. Listen to discovery events.
+ ctx.event().addLocalEventListener(discoLsnr, EVT_NODE_FAILED,
EVT_NODE_LEFT);
+
+ // 5. Check whether node has left before disco listener has been
installed.
+ taskNode = ctx.discovery().node(taskNodeId);
+
+ if (taskNode == null)
+ throw new IgniteCheckedException("Node that originated task
execution has left grid: " + taskNodeId);
+
+ // 6. Wait for result.
+ lock.lock();
+
+ try {
+ long netTimeout = ctx.config().getNetworkTimeout();
+
+ if (t.isEmpty())
+ cond.await(netTimeout, MILLISECONDS);
+
+ if (t.isEmpty())
+ throw new IgniteCheckedException("Timed out waiting for
job siblings (consider increasing" +
+ "'networkTimeout' configuration property) [ses=" +
ses + ", netTimeout=" + netTimeout + ']');
+
+ // Error is set?
+ if (t.get1() != null)
+ throw new IgniteCheckedException(t.get1());
+ else
+ // Return result
+ return t.get2().jobSiblings();
+ }
+ catch (InterruptedException e) {
+ throw new IgniteCheckedException("Interrupted while waiting
for job siblings response: " + ses, e);
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+ finally {
+ ctx.io().removeMessageListener(topic, msgLsnr);
+ ctx.event().removeLocalEventListener(discoLsnr);
+ }
+ }
+
+ /**
+ * Notify processor that master leave aware handler must be invoked on
all jobs with the given session ID.
+ *
+ * @param sesId Session ID.
+ */
+ public void masterLeaveLocal(IgniteUuid sesId) {
+ assert sesId != null;
+
+ for (GridJobWorker job : activeJobs.values())
+ if (job.getSession().getId().equals(sesId))
+ job.onMasterNodeLeft();
+ }
+
+ /**
+ * @param sesId Session ID.
+ * @param jobId Job ID.
+ * @param sys System flag.
+ */
+ public void cancelJob(@Nullable final IgniteUuid sesId, @Nullable final
IgniteUuid jobId, final boolean sys) {
+ assert sesId != null || jobId != null;
+
+ rwLock.readLock();
+
+ try {
+ if (stopping && cancelOnStop) {
+ if (log.isDebugEnabled())
+ log.debug("Received job cancellation request while
stopping grid with cancellation " +
+ "(will ignore) [sesId=" + sesId + ", jobId=" + jobId
+ ", sys=" + sys + ']');
+
+ return;
+ }
+
+ // Put either job ID or session ID (they are unique).
+ cancelReqs.putIfAbsent(jobId != null ? jobId : sesId, sys);
+
+ IgnitePredicate<GridJobWorker> idsMatch = new P1<GridJobWorker>()
{
+ @Override public boolean apply(GridJobWorker e) {
+ return sesId != null ?
+ jobId != null ?
+ e.getSession().getId().equals(sesId) &&
e.getJobId().equals(jobId) :
+ e.getSession().getId().equals(sesId) :
+ e.getJobId().equals(jobId);
+ }
+ };
+
+ // If we don't have jobId then we have to iterate
+ if (jobId == null) {
+ if (!jobAlwaysActivate) {
+ // If job gets removed from passive jobs it never gets
activated.
+ F.forEach(passiveJobs.values(), new CI1<GridJobWorker>() {
+ @Override public void apply(GridJobWorker job) {
+ cancelPassiveJob(job);
+ }
+ }, idsMatch);
+ }
+
+ F.forEach(activeJobs.values(), new CI1<GridJobWorker>() {
+ @Override public void apply(GridJobWorker job) {
+ cancelActiveJob(job, sys);
+ }
+ }, idsMatch);
+ }
+ else {
+ if (!jobAlwaysActivate) {
+ GridJobWorker passiveJob = passiveJobs.get(jobId);
+
+ if (passiveJob != null && idsMatch.apply(passiveJob) &&
cancelPassiveJob(passiveJob))
+ return;
+ }
+
+ GridJobWorker activeJob = activeJobs.get(jobId);
+
+ if (activeJob != null && idsMatch.apply(activeJob))
+ cancelActiveJob(activeJob, sys);
+ }
+ }
+ finally {
+ rwLock.readUnlock();
+ }
+ }
+
+ /**
+ * Tries to cancel passive job. No-op if job is not in 'passive' state.
+ *
+ * @param job Job to cancel.
+ * @return {@code True} if succeeded.
+ */
+ private boolean cancelPassiveJob(GridJobWorker job) {
+ assert !jobAlwaysActivate;
+
+ if (passiveJobs.remove(job.getJobId(), job)) {
+ if (log.isDebugEnabled())
+ log.debug("Job has been cancelled before activation: " + job);
+
+ canceledJobsCnt.increment();
+
+ return true;
+ }
+
+ return false;
+ }
+
+ /**
+ * Tries to cancel active job. No-op if job is not in 'active' state.
+ *
+ * @param job Job to cancel.
+ * @param sys Flag indicating whether this is a system cancel.
+ */
+ private void cancelActiveJob(GridJobWorker job, boolean sys) {
+ if (activeJobs.remove(job.getJobId(), job)) {
+ cancelledJobs.put(job.getJobId(), job);
+
+ if (finishedJobs.contains(job.getJobId()))
+ // Job has finished concurrently.
+ cancelledJobs.remove(job.getJobId(), job);
+ else
+ // No reply, since it is not cancel from collision.
+ cancelJob(job, sys);
+ }
+ }
+
+ /**
+ * Handles collisions.
+ * <p>
+ * In most cases this method should be called from main read lock
+ * to avoid jobs activation after node stop has started.
+ */
+ private void handleCollisions() {
+ assert !jobAlwaysActivate;
+
+ if (handlingCollision.get()) {
+ if (log.isDebugEnabled())
+ log.debug("Skipping recursive collision handling.");
+
+ return;
+ }
+
+ handlingCollision.set(Boolean.TRUE);
+
+ try {
+ if (log.isDebugEnabled())
+ log.debug("Before handling collisions.");
+
+ // Invoke collision SPI.
+ ctx.collision().onCollision(
+ // Passive jobs view.
+ new
AbstractCollection<org.apache.ignite.spi.collision.CollisionJobContext>() {
+ @NotNull @Override public
Iterator<org.apache.ignite.spi.collision.CollisionJobContext> iterator() {
+ final Iterator<GridJobWorker> iter =
passiveJobs.values().iterator();
+
+ return new
Iterator<org.apache.ignite.spi.collision.CollisionJobContext>() {
+ @Override public boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ @Override public
org.apache.ignite.spi.collision.CollisionJobContext next() {
+ return new CollisionJobContext(iter.next(),
true);
+ }
+
+ @Override public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+
+ @Override public int size() {
+ return passiveJobs.size();
+ }
+ },
+
+ // Active jobs view.
+ new
AbstractCollection<org.apache.ignite.spi.collision.CollisionJobContext>() {
+ @NotNull @Override public
Iterator<org.apache.ignite.spi.collision.CollisionJobContext> iterator() {
+ final Iterator<GridJobWorker> iter =
activeJobs.values().iterator();
+
+ return new
Iterator<org.apache.ignite.spi.collision.CollisionJobContext>() {
+ private GridJobWorker w;
+
+ {
+ advance();
+ }
+
+ /**
+ *
+ */
+ void advance() {
+ assert w == null;
+
+ while(iter.hasNext()) {
+ GridJobWorker w0 = iter.next();
+
+ assert !w0.isInternal();
+
+ if (!w0.held()) {
+ w = w0;
+
+ break;
+ }
+ }
+ }
+
+ @Override public boolean hasNext() {
+ return w != null;
+ }
+
+ @Override public
org.apache.ignite.spi.collision.CollisionJobContext next() {
+ if (w == null)
+ throw new NoSuchElementException();
+
+
org.apache.ignite.spi.collision.CollisionJobContext ret = new
CollisionJobContext(w, false);
+
+ w = null;
+
+ advance();
+
+ return ret;
+ }
+
+ @Override public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+
+ @Override public int size() {
+ int ret = activeJobs.size() - heldJobs.size();
+
+ return ret > 0 ? ret : 0;
+ }
+ },
+
+ // Held jobs view.
+ new
AbstractCollection<org.apache.ignite.spi.collision.CollisionJobContext>() {
+ @NotNull @Override public
Iterator<org.apache.ignite.spi.collision.CollisionJobContext> iterator() {
+ final Iterator<GridJobWorker> iter =
activeJobs.values().iterator();
+
+ return new
Iterator<org.apache.ignite.spi.collision.CollisionJobContext>() {
+ private GridJobWorker w;
+
+ {
+ advance();
+ }
+
+ /**
+ *
+ */
+ void advance() {
+ assert w == null;
+
+ while(iter.hasNext()) {
+ GridJobWorker w0 = iter.next();
+
+ assert !w0.isInternal();
+
+ if (w0.held()) {
+ w = w0;
+
+ break;
+ }
+ }
+ }
+
+ @Override public boolean hasNext() {
+ return w != null;
+ }
+
+ @Override public
org.apache.ignite.spi.collision.CollisionJobContext next() {
+ if (w == null)
+ throw new NoSuchElementException();
+
+
org.apache.ignite.spi.collision.CollisionJobContext ret = new
CollisionJobContext(w, false);
+
+ w = null;
+
+ advance();
+
+ return ret;
+ }
+
+ @Override public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+
+ @Override public int size() {
+ return heldJobs.size();
+ }
+ });
+
+ if (metricsUpdateFreq > -1L)
+ updateJobMetrics();
+ }
+ finally {
+ handlingCollision.set(Boolean.FALSE);
+ }
+ }
+
+ /**
+ *
+ */
+ private void updateJobMetrics() {
+ assert metricsUpdateFreq > -1L;
+
+ if (metricsUpdateFreq == 0L)
+ updateJobMetrics0();
+ else {
+ long now = U.currentTimeMillis();
+ long lastUpdate = metricsLastUpdateTstamp.get();
+
+ if (now - lastUpdate > metricsUpdateFreq &&
metricsLastUpdateTstamp.compareAndSet(lastUpdate, now))
+ updateJobMetrics0();
+ }
+ }
+
+ /**
+ *
+ */
+ private void updateJobMetrics0() {
+ assert metricsUpdateFreq > -1L;
+
+ GridJobMetricsSnapshot m = new GridJobMetricsSnapshot();
+
+ m.setRejectJobs((int)rejectedJobsCnt.sumThenReset());
+ m.setStartedJobs((int)startedJobsCnt.sumThenReset());
+
+ // Iterate over active jobs to determine max execution time.
+ int cnt = 0;
+
+ for (GridJobWorker jobWorker : activeJobs.values()) {
+ assert !jobWorker.isInternal();
+
+ cnt++;
+
+ if (!jobWorker.held()) {
+ long execTime = jobWorker.getExecuteTime();
+
+ if (execTime > m.getMaximumExecutionTime())
+ m.setMaximumExecutionTime(execTime);
+ }
+ }
+
+ m.setActiveJobs(cnt);
+
+ cnt = 0;
+
+ // Do this only if collision SPI is used. Otherwise, 0 is correct
value
+ // for passive jobs count and max wait time.
+ if (!jobAlwaysActivate) {
+ // Iterate over passive jobs to determine max queued time.
+ for (GridJobWorker jobWorker : passiveJobs.values()) {
+ // We don't expect that there are any passive internal jobs.
+ assert !jobWorker.isInternal();
+
+ cnt++;
+
+ long queuedTime = jobWorker.getQueuedTime();
+
+ if (queuedTime > m.getMaximumWaitTime())
+ m.setMaximumWaitTime(queuedTime);
+
+ m.setWaitTime(m.getWaitTime() + jobWorker.getQueuedTime());
+ }
+
+ m.setPassiveJobs(cnt);
+ }
+
+ m.setFinishedJobs((int)finishedJobsCnt.sumThenReset());
+ m.setExecutionTime(finishedJobsTime.sumThenReset());
+ m.setCancelJobs((int)canceledJobsCnt.sumThenReset());
+
+ long maxFinishedTime = maxFinishedJobsTime.getAndSet(0);
+
+ if (maxFinishedTime > m.getMaximumExecutionTime())
+ m.setMaximumExecutionTime(maxFinishedTime);
+
+ // CPU load.
+ m.setCpuLoad(ctx.discovery().metrics().getCurrentCpuLoad());
+
+ ctx.jobMetric().addSnapshot(m);
+ }
+
+ /**
+ * @param node Node.
+ * @param req Request.
+ */
+ @SuppressWarnings("TooBroadScope")
+ public void processJobExecuteRequest(ClusterNode node, final
GridJobExecuteRequest req) {
+ if (log.isDebugEnabled())
+ log.debug("Received job request message [req=" + req + ",
nodeId=" + node.id() + ']');
+
+ GridJobWorker job = null;
+
+ rwLock.readLock();
+
+ try {
+ if (stopping) {
+ if (log.isDebugEnabled())
+ log.debug("Received job execution request while stopping
this node (will ignore): " + req);
+
+ return;
+ }
+
+ long endTime = req.getCreateTime() + req.getTimeout();
+
+ // Account for overflow.
+ if (endTime < 0)
+ endTime = Long.MAX_VALUE;
+
+ GridDeployment tmpDep = req.isForceLocalDeployment() ?
+ ctx.deploy().getLocalDeployment(req.getTaskClassName()) :
+ ctx.deploy().getGlobalDeployment(
+ req.getDeploymentMode(),
+ req.getTaskName(),
+ req.getTaskClassName(),
+ req.getUserVersion(),
+ node.id(),
+ req.getClassLoaderId(),
+ req.getLoaderParticipants(),
+ null);
+
+ if (tmpDep == null) {
+ if (log.isDebugEnabled())
+ log.debug("Checking local tasks...");
+
+ // Check local tasks.
+ for (Map.Entry<String, GridDeployment> d :
ctx.task().getUsedDeploymentMap().entrySet()) {
+ if
(d.getValue().classLoaderId().equals(req.getClassLoaderId())) {
+ assert d.getValue().local();
+
+ tmpDep = d.getValue();
+
+ break;
+ }
+ }
+ }
+
+ final GridDeployment dep = tmpDep;
+
+ if (log.isDebugEnabled())
+ log.debug("Deployment: " + dep);
+
+ boolean releaseDep = true;
+
+ try {
+ if (dep != null && dep.acquire()) {
+ GridJobSessionImpl jobSes;
+ GridJobContextImpl jobCtx;
+
+ try {
+ List<ComputeJobSibling> siblings = null;
+
+ if (!req.isDynamicSiblings()) {
+ Collection<ComputeJobSibling> siblings0 =
req.getSiblings();
+
+ if (siblings0 == null) {
+ assert req.getSiblingsBytes() != null;
+
+ siblings0 =
marsh.unmarshal(req.getSiblingsBytes(), null);
+ }
+
+ siblings = new ArrayList<>(siblings0);
+ }
+
+ Map<Object, Object> sesAttrs = null;
+
+ if (req.isSessionFullSupport()) {
+ sesAttrs = req.getSessionAttributes();
+
+ if (sesAttrs == null)
+ sesAttrs =
marsh.unmarshal(req.getSessionAttributesBytes(),
+ dep.classLoader());
+ }
+
+ // Note that we unmarshal session/job attributes here
with proper class loader.
+ GridTaskSessionImpl taskSes =
ctx.session().createTaskSession(
+ req.getSessionId(),
+ node.id(),
+ req.getTaskName(),
+ dep,
+ req.getTaskClassName(),
+ req.topology(),
+ req.getStartTaskTime(),
+ endTime,
+ siblings,
+ sesAttrs,
+ req.isSessionFullSupport(),
+ req.getSubjectId());
+
+ taskSes.setCheckpointSpi(req.getCheckpointSpi());
+ taskSes.setClassLoader(dep.classLoader());
+
+ jobSes = new GridJobSessionImpl(ctx, taskSes,
req.getJobId());
+
+ Map<? extends Serializable, ? extends Serializable>
jobAttrs = req.getJobAttributes();
+
+ if (jobAttrs == null)
+ jobAttrs =
marsh.unmarshal(req.getJobAttributesBytes(), dep.classLoader());
+
+ jobCtx = new GridJobContextImpl(ctx, req.getJobId(),
jobAttrs);
+ }
+ catch (IgniteCheckedException e) {
+ IgniteCheckedException ex = new
IgniteCheckedException("Failed to deserialize task attributes [taskName=" +
+ req.getTaskName() + ", taskClsName=" +
req.getTaskClassName() + ", codeVer=" +
+ req.getUserVersion() + ", taskClsLdr=" +
dep.classLoader() + ']', e);
+
+ U.error(log, ex.getMessage(), e);
+
+ handleException(node, req, ex, endTime);
+
+ return;
+ }
+
+ job = new GridJobWorker(
+ ctx,
+ dep,
+ req.getCreateTime(),
+ jobSes,
+ jobCtx,
+ req.getJobBytes(),
+ req.getJob(),
+ node,
+ req.isInternal(),
+ evtLsnr,
+ holdLsnr);
+
+ jobCtx.job(job);
+
+ // If exception occurs on job initialization, deployment
is released in job listener.
+ releaseDep = false;
+
+ if (job.initialize(dep,
dep.deployedClass(req.getTaskClassName()))) {
+ // Internal jobs will always be executed
synchronously.
+ if (job.isInternal()) {
+ // This is an internal job and can be executed
inside busy lock
+ // since job is expected to be short.
+ // This is essential for proper stop without
races.
+ job.run();
+
+ // No execution outside lock.
+ job = null;
+ }
+ else if (jobAlwaysActivate) {
+ if (onBeforeActivateJob(job)) {
+ if (ctx.localNodeId().equals(node.id())) {
+ // Always execute in another thread for
local node.
+ executeAsync(job);
+
+ // No sync execution.
+ job = null;
+ }
+ else if (metricsUpdateFreq > -1L)
+ // Job will be executed synchronously.
+ startedJobsCnt.increment();
+ }
+ else
+ // Job has been cancelled.
+ // Set to null, to avoid sync execution.
+ job = null;
+ }
+ else {
+ GridJobWorker old =
passiveJobs.putIfAbsent(job.getJobId(), job);
+
+ if (old == null)
+ handleCollisions();
+ else
+ U.error(log, "Received computation request
with duplicate job ID (could be " +
+ "network malfunction, source node may
hang if task timeout was not set) " +
+ "[srcNode=" + node.id() +
+ ", jobId=" + req.getJobId() + ", sesId="
+ req.getSessionId() +
+ ", locNodeId=" + ctx.localNodeId() + ']');
+
+ // No sync execution.
+ job = null;
+ }
+ }
+ else
+ // Job was not initialized, no execution.
+ job = null;
+ }
+ else {
+ // Deployment is null.
+ IgniteCheckedException ex = new
IgniteDeploymentException("Task was not deployed or was redeployed since " +
+ "task execution [taskName=" + req.getTaskName() + ",
taskClsName=" + req.getTaskClassName() +
+ ", codeVer=" + req.getUserVersion() + ", clsLdrId=" +
req.getClassLoaderId() +
+ ", seqNum=" + req.getClassLoaderId().localId() + ",
depMode=" + req.getDeploymentMode() +
+ ", dep=" + dep + ']');
+
+ U.error(log, ex.getMessage(), ex);
+
+ handleException(node, req, ex, endTime);
+ }
+ }
+ finally {
+ if (dep != null && releaseDep)
+ release(dep);
+ }
+ }
+ finally {
+ rwLock.readUnlock();
+ }
+
+ if (job != null)
+ job.run();
+ }
+
+ /**
+ * Callback from job worker to set current task session for execution.
+ *
+ * @param ses Session.
+ */
+ public void currentTaskSession(ComputeTaskSession ses) {
+ currentSess.set(ses);
+ }
+
+ /**
+ * Gets hash of task name executed by current thread.
+ *
+ * @return Task name hash or {@code 0} if security is disabled.
+ */
+ public int currentTaskNameHash() {
+ String name = currentTaskName();
+
+ return name == null ? 0 : name.hashCode();
+ }
+
+ /**
+ * Gets name task executed by current thread.
+ *
+ * @return Task name or {@code null} if security is disabled.
+ */
+ public String currentTaskName() {
+ if (!ctx.security().enabled())
+ return null;
+
+ ComputeTaskSession ses = currentSess.get();
+
+ if (ses == null)
+ return null;
+
+ return ses.getTaskName();
+ }
+
+ /**
+ * @param jobWorker Worker.
+ * @return {@code True} if job has not been cancelled and should be
activated.
+ */
+ private boolean onBeforeActivateJob(GridJobWorker jobWorker) {
+ assert jobWorker != null;
+
+ activeJobs.put(jobWorker.getJobId(), jobWorker);
+
+ // Check if job has been concurrently cancelled.
+ Boolean sysCancelled = cancelReqs.get(jobWorker.getSession().getId());
+
+ if (sysCancelled == null)
+ sysCancelled = cancelReqs.get(jobWorker.getJobId());
+
+ if (sysCancelled != null) {
+ // Job has been concurrently cancelled.
+ // Remove from active jobs.
+ activeJobs.remove(jobWorker.getJobId(), jobWorker);
+
+ // Even if job has been removed from another thread, we need to
reject it
+ // here since job has never been executed.
+ IgniteCheckedException e2 = new ComputeExecutionRejectedException(
+ "Job was cancelled before execution [jobSes=" + jobWorker.
+ getSession() + ", job=" + jobWorker.getJob() + ']');
+
+ jobWorker.finishJob(null, e2, !sysCancelled);
+
+ return false;
+ }
+
+ // Job has not been cancelled and should be activated.
+ // However we need to check if master is alive before job will get
+ // its runner thread for proper master leave handling.
+ if (ctx.discovery().node(jobWorker.getTaskNode().id()) == null &&
+ activeJobs.remove(jobWorker.getJobId(), jobWorker)) {
+ // Add to cancelled jobs.
+ cancelledJobs.put(jobWorker.getJobId(), jobWorker);
+
+ if (!jobWorker.onMasterNodeLeft()) {
+ U.warn(log, "Job is being cancelled because master task node
left grid " +
+ "(as there is no one waiting for results, job will not be
failed over): " +
+ jobWorker.getJobId());
+
+ cancelJob(jobWorker, true);
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * @param jobWorker Job worker.
+ * @return {@code True} if job has been submitted to pool.
+ */
+ private boolean executeAsync(GridJobWorker jobWorker) {
+ try {
+ ctx.config().getExecutorService().execute(jobWorker);
+
+ if (metricsUpdateFreq > -1L)
+ startedJobsCnt.increment();
+
+ return true;
+ }
+ catch (RejectedExecutionException e) {
+ // Remove from active jobs.
+ activeJobs.remove(jobWorker.getJobId(), jobWorker);
+
+ // Even if job was removed from another thread, we need to reject
it
+ // here since job has never been executed.
+ IgniteCheckedException e2 = new
ComputeExecutionRejectedException("Job has been rejected " +
+ "[jobSes=" + jobWorker.getSession() + ", job=" +
jobWorker.getJob() + ']', e);
+
+ if (metricsUpdateFreq > -1L)
+ rejectedJobsCnt.increment();
+
+ jobWorker.finishJob(null, e2, true);
+ }
+
+ return false;
+ }
+
+ /**
+ * Handles errors that happened prior to job creation.
+ *
+ * @param node Sender node.
+ * @param req Job execution request.
+ * @param ex Exception that happened.
+ * @param endTime Job end time.
+ */
+ private void handleException(ClusterNode node, GridJobExecuteRequest req,
IgniteCheckedException ex, long endTime) {
+ UUID locNodeId = ctx.localNodeId();
+
+ ClusterNode sndNode = ctx.discovery().node(node.id());
+
+ if (sndNode == null) {
+ U.warn(log, "Failed to reply to sender node because it left grid
[nodeId=" + node.id() +
+ ", jobId=" + req.getJobId() + ']');
+
+ if (ctx.event().isRecordable(EVT_JOB_FAILED)) {
+ IgniteJobEvent evt = new IgniteJobEvent();
+
+ evt.jobId(req.getJobId());
+ evt.message("Job reply failed (original task node left grid):
" + req.getJobId());
+ evt.node(ctx.discovery().localNode());
+ evt.taskName(req.getTaskName());
+ evt.taskClassName(req.getTaskClassName());
+ evt.taskSessionId(req.getSessionId());
+ evt.type(EVT_JOB_FAILED);
+ evt.taskNode(node);
+ evt.taskSubjectId(req.getSubjectId());
+
+ // Record job reply failure.
+ ctx.event().record(evt);
+ }
+
+ return;
+ }
+
+ try {
+ boolean loc = ctx.localNodeId().equals(sndNode.id()) &&
!ctx.config().isMarshalLocalJobs();
+
+ GridJobExecuteResponse jobRes = new GridJobExecuteResponse(
+ locNodeId,
+ req.getSessionId(),
+ req.getJobId(),
+ loc ? null : marsh.marshal(ex),
+ ex,
+ loc ? null : marsh.marshal(null),
+ null,
+ loc ? null : marsh.marshal(null),
+ null,
+ false);
+
+ if (req.isSessionFullSupport()) {
+ // Send response to designated job topic.
+ // Always go through communication to preserve order,
+ // if attributes are enabled.
+ // Job response topic.
+ Object topic = TOPIC_TASK.topic(req.getJobId(), locNodeId);
+
+ long timeout = endTime - U.currentTimeMillis();
+
+ if (timeout <= 0)
+ // Ignore the actual timeout and send response anyway.
+ timeout = 1;
+
+ // Send response to designated job topic.
+ // Always go through communication to preserve order.
- long msgId = ctx.io().nextMessageId(topic, sndNode.id());
-
- ctx.io().removeMessageId(topic);
-
+ ctx.io().sendOrderedMessage(
+ sndNode,
+ topic,
- msgId,
+ jobRes,
+ req.isInternal() ? MANAGEMENT_POOL : SYSTEM_POOL,
+ timeout,
+ false);
+ }
+ else if (ctx.localNodeId().equals(sndNode.id()))
+ ctx.task().processJobExecuteResponse(ctx.localNodeId(),
jobRes);
+ else
+ // Send response to common topic as unordered message.
+ ctx.io().send(sndNode, TOPIC_TASK, jobRes, req.isInternal() ?
MANAGEMENT_POOL : SYSTEM_POOL);
+ }
+ catch (IgniteCheckedException e) {
+ // The only option here is to log, as we must assume that
resending will fail too.
+ if (isDeadNode(node.id()))
+ // Avoid stack trace for left nodes.
+ U.error(log, "Failed to reply to sender node because it left
grid [nodeId=" + node.id() +
+ ", jobId=" + req.getJobId() + ']');
+ else {
+ assert sndNode != null;
+
+ U.error(log, "Error sending reply for job [nodeId=" +
sndNode.id() + ", jobId=" +
+ req.getJobId() + ']', e);
+ }
+
+ if (ctx.event().isRecordable(EVT_JOB_FAILED)) {
+ IgniteJobEvent evt = new IgniteJobEvent();
+
+ evt.jobId(req.getJobId());
+ evt.message("Failed to send reply for job: " +
req.getJobId());
+ evt.node(ctx.discovery().localNode());
+ evt.taskName(req.getTaskName());
+ evt.taskClassName(req.getTaskClassName());
+ evt.taskSessionId(req.getSessionId());
+ evt.type(EVT_JOB_FAILED);
+ evt.taskNode(node);
+ evt.taskSubjectId(req.getSubjectId());
+
+ // Record job reply failure.
+ ctx.event().record(evt);
+ }
+ }
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param req Request.
+ */
+ @SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter",
"RedundantCast"})
+ private void processTaskSessionRequest(UUID nodeId,
GridTaskSessionRequest req) {
+ rwLock.readLock();
+
+ try {
+ if (stopping) {
+ if (log.isDebugEnabled())
+ log.debug("Received job session request while stopping
grid (will ignore): " + req);
+
+ return;
+ }
+
+ GridTaskSessionImpl ses =
ctx.session().getSession(req.getSessionId());
+
+ if (ses == null) {
+ if (log.isDebugEnabled())
+ log.debug("Received job session request for non-existing
session: " + req);
+
+ return;
+ }
+
+ boolean loc = ctx.localNodeId().equals(nodeId) &&
!ctx.config().isMarshalLocalJobs();
+
+ Map<?, ?> attrs = loc ? req.getAttributes() :
+ (Map<?, ?>)marsh.unmarshal(req.getAttributesBytes(),
ses.getClassLoader());
+
+ if (ctx.event().isRecordable(EVT_TASK_SESSION_ATTR_SET)) {
+ IgniteEvent evt = new IgniteTaskEvent(
+ ctx.discovery().localNode(),
+ "Changed attributes: " + attrs,
+ EVT_TASK_SESSION_ATTR_SET,
+ ses.getId(),
+ ses.getTaskName(),
+ ses.getTaskClassName(),
+ false,
+ null);
+
+ ctx.event().record(evt);
+ }
+
+ synchronized (ses) {
+ ses.setInternal(attrs);
+ }
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to deserialize session attributes.", e);
+ }
+ finally {
+ rwLock.readUnlock();
+ }
+ }
+
+ /**
+ * 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} */
+ @Override public void printMemoryStats() {
+ X.println(">>>");
+ X.println(">>> Job processor memory stats [grid=" + ctx.gridName() +
']');
+ X.println(">>> activeJobsSize: " + activeJobs.size());
+ X.println(">>> passiveJobsSize: " + (jobAlwaysActivate ? "n/a" :
passiveJobs.size()));
+ X.println(">>> cancelledJobsSize: " + cancelledJobs.size());
+ X.println(">>> cancelReqsSize: " + cancelReqs.sizex());
+ X.println(">>> finishedJobsSize: " + finishedJobs.sizex());
+ }
+
+ /**
+ *
+ */
+ private class CollisionJobContext extends GridCollisionJobContextAdapter {
+ /** */
+ private final boolean passive;
+
+ /**
+ * @param jobWorker Job Worker.
+ * @param passive {@code True} if job is on waiting list on creation
time.
+ */
+ CollisionJobContext(GridJobWorker jobWorker, boolean passive) {
+ super(jobWorker);
+
+ assert !jobWorker.isInternal();
+ assert !jobAlwaysActivate;
+
+ this.passive = passive;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean activate() {
+ GridJobWorker jobWorker = getJobWorker();
+
+ return passiveJobs.remove(jobWorker.getJobId(), jobWorker) &&
+ onBeforeActivateJob(jobWorker) &&
+ executeAsync(jobWorker);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean cancel() {
+ GridJobWorker jobWorker = getJobWorker();
+
+ cancelReqs.putIfAbsent(jobWorker.getJobId(), false);
+
+ boolean ret = false;
+
+ if (passive) {
+ // If waiting job being rejected.
+ if (passiveJobs.remove(jobWorker.getJobId(), jobWorker)) {
+ rejectJob(jobWorker, true);
+
+ if (metricsUpdateFreq > -1L)
+ rejectedJobsCnt.increment();
+
+ ret = true;
+ }
+ }
+ // If active job being cancelled.
+ else if (activeJobs.remove(jobWorker.getJobId(), jobWorker)) {
+ cancelledJobs.put(jobWorker.getJobId(), jobWorker);
+
+ if (finishedJobs.contains(jobWorker.getJobId()))
+ // Job has finished concurrently.
+ cancelledJobs.remove(jobWorker.getJobId(), jobWorker);
+ else
+ // We do apply cancel as many times as user cancel job.
+ cancelJob(jobWorker, false);
+
+ ret = true;
+ }
+
+ return ret;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(CollisionJobContext.class, this);
+ }
+ }
+
+ /**
+ *
+ */
+ private class CollisionExternalListener implements
org.apache.ignite.spi.collision.CollisionExternalListener {
+ /** {@inheritDoc} */
+ @Override public void onExternalCollision() {
+ assert !jobAlwaysActivate;
+
+ if (log.isDebugEnabled())
+ log.debug("Received external collision event.");
+
+ rwLock.readLock();
+
+ try {
+ if (stopping) {
+ if (log.isDebugEnabled())
+ log.debug("Received external collision notification
while stopping grid (will ignore).");
+
+ return;
+ }
+
+ handleCollisions();
+ }
+ finally {
+ rwLock.readUnlock();
+ }
+ }
+ }
+
+ /**
+ * Handles job state changes.
+ */
+ private class JobEventListener implements GridJobEventListener {
+ /** */
+ private final GridMessageListener sesLsnr = new JobSessionListener();
+
+ /** {@inheritDoc} */
+ @Override public void onJobStarted(GridJobWorker worker) {
+ if (log.isDebugEnabled())
+ log.debug("Received onJobStarted() callback: " + worker);
+
+ if (metricsUpdateFreq > -1L)
+ updateJobMetrics();
+
+ // Register for timeout notifications.
+ if (worker.endTime() < Long.MAX_VALUE)
+ ctx.timeout().addTimeoutObject(worker);
+
+ if (worker.getSession().isFullSupport())
+ // Register session request listener for this job.
+ ctx.io().addMessageListener(worker.getJobTopic(), sesLsnr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onBeforeJobResponseSent(GridJobWorker worker) {
+ if (log.isDebugEnabled())
+ log.debug("Received onBeforeJobResponseSent() callback: " +
worker);
+
+ assert jobAlwaysActivate ||
!passiveJobs.containsKey(worker.getJobId());
+
+ if (worker.getSession().isFullSupport()) {
+ // Unregister session request listener for this jobs.
+ ctx.io().removeMessageListener(worker.getJobTopic());
-
- // Unregister message IDs used for sending.
- ctx.io().removeMessageId(worker.getTaskTopic());
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onJobFinished(GridJobWorker worker) {
+ if (log.isDebugEnabled())
+ log.debug("Received onJobFinished() callback: " + worker);
+
+ GridJobSessionImpl ses = worker.getSession();
+
+ // If last job for the task on this node.
+ if (ses.isFullSupport() &&
ctx.session().removeSession(ses.getId())) {
+ ses.onClosed();
+
+ // Unregister checkpoints.
+ ctx.checkpoint().onSessionEnd(ses, true);
+ }
+
+ // Unregister from timeout notifications.
+ if (worker.endTime() < Long.MAX_VALUE)
+ ctx.timeout().removeTimeoutObject(worker);
+
+ release(worker.getDeployment());
+
+ finishedJobs.add(worker.getJobId());
+
+ if (!worker.isInternal()) {
+ // Increment job execution counter. This counter gets
+ // reset once this job will be accounted for in metrics.
+ finishedJobsCnt.increment();
+
+ // Increment job execution time. This counter gets
+ // reset once this job will be accounted for in metrics.
+ long execTime = worker.getExecuteTime();
+
+ finishedJobsTime.add(execTime);
+
+ maxFinishedJobsTime.setIfGreater(execTime);
+
+ if (jobAlwaysActivate) {
+ if (metricsUpdateFreq > -1L)
+ updateJobMetrics();
+ }
+ else {
+ rwLock.readLock();
+
+ try {
+ if (stopping) {
+ if (log.isDebugEnabled())
+ log.debug("Skipping collision handling on job
finish (node is stopping).");
+
+ return;
+ }
+
+ handleCollisions();
+ }
+ finally {
+ rwLock.readUnlock();
+ }
+ }
+
+ if (!activeJobs.remove(worker.getJobId(), worker))
+ cancelledJobs.remove(worker.getJobId(), worker);
+
+ heldJobs.remove(worker.getJobId());
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ private class JobHoldListener implements GridJobHoldListener {
+ /** {@inheritDoc} */
+ @Override public void onHold(GridJobWorker worker) {
+ if (log.isDebugEnabled())
+ log.debug("Received onHold() callback [worker=" + worker +
']');
+
+ if (activeJobs.containsKey(worker.getJobId())) {
+ heldJobs.add(worker.getJobId());
+
+ if (!activeJobs.containsKey(worker.getJobId()))
+ heldJobs.remove(worker.getJobId());
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUnhold(GridJobWorker worker) {
+ if (log.isDebugEnabled())
+ log.debug("Received onUnhold() callback [worker=" + worker +
", active=" + activeJobs +
+ ", held=" + heldJobs + ']');
+
+ heldJobs.remove(worker.getJobId());
+ }
+ }
+
+ /**
+ *
+ */
+ private class JobSessionListener implements GridMessageListener {
+ /** {@inheritDoc} */
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ assert nodeId != null;
+ assert msg != null;
+
+ if (log.isDebugEnabled())
+ log.debug("Received session attribute request message [msg="
+ msg + ", nodeId=" + nodeId + ']');
+
+ processTaskSessionRequest(nodeId, (GridTaskSessionRequest)msg);
+ }
+ }
+
+ /**
+ * Handles task and job cancellations.
+ */
+ private class JobCancelListener implements GridMessageListener {
+ /** {@inheritDoc} */
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ assert nodeId != null;
+ assert msg != null;
+
+ GridJobCancelRequest cancelMsg = (GridJobCancelRequest)msg;
+
+ if (log.isDebugEnabled())
+ log.debug("Received job cancel request [cancelMsg=" +
cancelMsg + ", nodeId=" + nodeId + ']');
+
+ cancelJob(cancelMsg.sessionId(), cancelMsg.jobId(),
cancelMsg.system());
+ }
+ }
+
+ /**
+ * Handles job execution requests.
+ */
+ private class JobExecutionListener implements GridMessageListener {
+ /** {@inheritDoc} */
+ @Override public void onMessage(UUID nodeId, Object msg) {
+ assert nodeId != null;
+ assert msg != null;
+
+ ClusterNode node = ctx.discovery().node(nodeId);
+
+ if (!ctx.discovery().alive(nodeId)) {
+ U.warn(log, "Received job request message from unknown node
(ignoring) " +
+ "[msg=" + msg + ", nodeId=" + nodeId + ']');
+
+ return;
+ }
+
+ assert node != null;
+
+ processJobExecuteRequest(node, (GridJobExecuteRequest)msg);
+ }
+ }
+
+ /**
+ * Listener to node discovery events.
+ */
+ private class JobDiscoveryListener implements GridLocalEventListener {
+ /**
+ * Counter used to determine whether all nodes updated metrics or not.
+ * This counter is reset every time collisions are handled.
+ */
+ private int metricsUpdateCntr;
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("fallthrough")
+ @Override public void onEvent(IgniteEvent evt) {
+ assert evt instanceof IgniteDiscoveryEvent;
+
+ boolean handleCollisions = false;
+
+ UUID nodeId = ((IgniteDiscoveryEvent)evt).eventNode().id();
+
+ // We should always process discovery events (even on stop,
+ // since we wait for jobs to complete if processor is stopped
+ // without cancellation).
+ switch (evt.type()) {
+ case EVT_NODE_LEFT:
+ case EVT_NODE_FAILED:
+ if (!jobAlwaysActivate) {
+ for (GridJobWorker job : passiveJobs.values()) {
+ if (job.getTaskNode().id().equals(nodeId)) {
+ if (passiveJobs.remove(job.getJobId(), job))
+ U.warn(log, "Task node left grid (job
will not be activated) " +
+ "[nodeId=" + nodeId + ", jobSes=" +
job.getSession() + ", job=" + job + ']');
+ }
+ }
+ }
+
+ for (GridJobWorker job : activeJobs.values()) {
+ if (job.getTaskNode().id().equals(nodeId) &&
!job.isFinishing() &&
+ activeJobs.remove(job.getJobId(), job)) {
+ // Add to cancelled jobs.
+ cancelledJobs.put(job.getJobId(), job);
+
+ if (finishedJobs.contains(job.getJobId()))
+ // Job has finished concurrently.
+ cancelledJobs.remove(job.getJobId(), job);
+ else if (!job.onMasterNodeLeft()) {
+ U.warn(log, "Job is being cancelled because
master task node left grid " +
+ "(as there is no one waiting for results,
job will not be failed over): " +
+ job.getJobId());
+
+ cancelJob(job, true);
+ }
+ }
+ }
+
+ handleCollisions = true;
+
+ break;
+
+ case EVT_NODE_METRICS_UPDATED:
+ // Check for less-than-equal rather than just equal
+ // in guard against topology changes.
+ if (ctx.discovery().allNodes().size() <=
++metricsUpdateCntr) {
+ metricsUpdateCntr = 0;
+
+ handleCollisions = true;
+ }
+
+ break;
+
+ default:
+ assert false;
+ }
+
+ if (handleCollisions) {
+ rwLock.readLock();
+
+ try {
+ if (stopping) {
+ if (log.isDebugEnabled())
+ log.debug("Skipped collision handling on
discovery event (node is stopping): " + evt);
+
+ return;
+ }
+
+ if (!jobAlwaysActivate)
+ handleCollisions();
+ else if (metricsUpdateFreq > -1L)
+ updateJobMetrics();
+ }
+ finally {
+ rwLock.readUnlock();
+ }
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ private class JobsMap extends ConcurrentLinkedHashMap<IgniteUuid,
GridJobWorker> {
+ /**
+ * @param initCap Initial capacity.
+ * @param loadFactor Load factor.
+ * @param concurLvl Concurrency level.
+ */
+ private JobsMap(int initCap, float loadFactor, int concurLvl) {
+ super(initCap, loadFactor, concurLvl);
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridJobWorker put(IgniteUuid key, GridJobWorker val)
{
+ assert !val.isInternal();
+
+ GridJobWorker old = super.put(key, val);
+
+ if (old != null)
+ U.warn(log, "Jobs map already contains mapping for key [key="
+ key + ", val=" + val +
+ ", old=" + old + ']');
+
+ return old;
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridJobWorker putIfAbsent(IgniteUuid key,
GridJobWorker val) {
+ assert !val.isInternal();
+
+ GridJobWorker old = super.putIfAbsent(key, val);
+
+ if (old != null)
+ U.warn(log, "Jobs map already contains mapping for key [key="
+ key + ", val=" + val +
+ ", old=" + old + ']');
+
+ return old;
+ }
+
+ /**
+ * @return Constant-time {@code size()}.
+ */
+ @Override public int size() {
+ return sizex();
+ }
+ }
+ }
