ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [29/50] [abbrv] 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 14:24:07 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplyPool.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplyPool.java
index 0000000,50cf8d9..865a16e
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplyPool.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplyPool.java
@@@ -1,0 -1,575 +1,567 @@@
+ /*
+  * 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.dht.preloader;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.internal.processors.cache.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.thread.*;
+ import org.apache.ignite.internal.managers.deployment.*;
+ import org.apache.ignite.internal.processors.cache.distributed.dht.*;
+ import org.apache.ignite.internal.util.lang.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ import org.apache.ignite.internal.util.worker.*;
+ import org.jetbrains.annotations.*;
+ 
+ import java.io.*;
+ import java.util.*;
+ import java.util.concurrent.*;
+ import java.util.concurrent.locks.*;
+ 
+ import static java.util.concurrent.TimeUnit.*;
+ import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.*;
+ 
+ /**
+  * Thread pool for supplying partitions to demanding nodes.
+  */
+ class GridDhtPartitionSupplyPool<K, V> {
+     /** */
+     private final GridCacheContext<K, V> cctx;
+ 
+     /** */
+     private final IgniteLogger log;
+ 
+     /** */
+     private final ReadWriteLock busyLock;
+ 
+     /** */
+     private GridDhtPartitionTopology<K, V> top;
+ 
+     /** */
+     private final Collection<SupplyWorker> workers = new LinkedList<>();
+ 
+     /** */
+     private final BlockingQueue<DemandMessage<K, V>> queue = new LinkedBlockingDeque<>();
+ 
+     /** */
+     private final boolean depEnabled;
+ 
+     /** Preload predicate. */
+     private IgnitePredicate<GridCacheEntryInfo<K, V>> preloadPred;
+ 
+     /**
+      * @param cctx Cache context.
+      * @param busyLock Shutdown lock.
+      */
+     GridDhtPartitionSupplyPool(GridCacheContext<K, V> cctx, ReadWriteLock busyLock) {
+         assert cctx != null;
+         assert busyLock != null;
+ 
+         this.cctx = cctx;
+         this.busyLock = busyLock;
+ 
+         log = cctx.logger(getClass());
+ 
+         top = cctx.dht().topology();
+ 
+         int poolSize = cctx.preloadEnabled() ? cctx.config().getPreloadThreadPoolSize() : 0;
+ 
+         for (int i = 0; i < poolSize; i++)
+             workers.add(new SupplyWorker());
+ 
+         cctx.io().addHandler(cctx.cacheId(), GridDhtPartitionDemandMessage.class, new CI2<UUID, GridDhtPartitionDemandMessage<K, V>>() {
+             @Override public void apply(UUID id, GridDhtPartitionDemandMessage<K, V> m) {
+                 processDemandMessage(id, m);
+             }
+         });
+ 
+         depEnabled = cctx.gridDeploy().enabled();
+     }
+ 
+     /**
+      *
+      */
+     void start() {
+         for (SupplyWorker w : workers)
+             new IgniteThread(cctx.gridName(), "preloader-supply-worker", w).start();
+     }
+ 
+     /**
+      *
+      */
+     void stop() {
+         U.cancel(workers);
+         U.join(workers, log);
+ 
+         top = null;
+     }
+ 
+     /**
+      * Sets preload predicate for supply pool.
+      *
+      * @param preloadPred Preload predicate.
+      */
+     void preloadPredicate(IgnitePredicate<GridCacheEntryInfo<K, V>> preloadPred) {
+         this.preloadPred = preloadPred;
+     }
+ 
+     /**
+      * @return Size of this thread pool.
+      */
+     int poolSize() {
+         return cctx.config().getPreloadThreadPoolSize();
+     }
+ 
+     /**
+      * @return {@code true} if entered to busy state.
+      */
+     private boolean enterBusy() {
+         if (busyLock.readLock().tryLock())
+             return true;
+ 
+         if (log.isDebugEnabled())
+             log.debug("Failed to enter to busy state (supplier is stopping): " + cctx.nodeId());
+ 
+         return false;
+     }
+ 
+     /**
+      * @param nodeId Sender node ID.
+      * @param d Message.
+      */
+     private void processDemandMessage(UUID nodeId, GridDhtPartitionDemandMessage<K, V> d) {
+         if (!enterBusy())
+             return;
+ 
+         try {
+             if (cctx.preloadEnabled()) {
+                 if (log.isDebugEnabled())
+                     log.debug("Received partition demand [node=" + nodeId + ", demand=" + d + ']');
+ 
+                 queue.offer(new DemandMessage<>(nodeId, d));
+             }
+             else
+                 U.warn(log, "Received partition demand message when preloading is disabled (will ignore): " + d);
+         }
+         finally {
+             leaveBusy();
+         }
+     }
+ 
+     /**
+      *
+      */
+     private void leaveBusy() {
+         busyLock.readLock().unlock();
+     }
+ 
+     /**
+      * @param deque Deque to poll from.
+      * @param w Worker.
+      * @return Polled item.
+      * @throws InterruptedException If interrupted.
+      */
+     @Nullable private <T> T poll(BlockingQueue<T> deque, GridWorker w) throws InterruptedException {
+         assert w != null;
+ 
+         // There is currently a case where {@code interrupted}
+         // flag on a thread gets flipped during stop which causes the pool to hang.  This check
+         // will always make sure that interrupted flag gets reset before going into wait conditions.
+         // The true fix should actually make sure that interrupted flag does not get reset or that
+         // interrupted exception gets propagated. Until we find a real fix, this method should
+         // always work to make sure that there is no hanging during stop.
+         if (w.isCancelled())
+             Thread.currentThread().interrupt();
+ 
+         return deque.poll(2000, MILLISECONDS);
+     }
+ 
+     /**
+      * Supply work.
+      */
+     private class SupplyWorker extends GridWorker {
+         /** Hide worker logger and use cache logger. */
+         private IgniteLogger log = GridDhtPartitionSupplyPool.this.log;
+ 
+         /**
+          * Default constructor.
+          */
+         private SupplyWorker() {
+             super(cctx.gridName(), "preloader-supply-worker", GridDhtPartitionSupplyPool.this.log);
+         }
+ 
+         /** {@inheritDoc} */
+         @Override protected void body() throws InterruptedException, IgniteInterruptedException {
+             while (!isCancelled()) {
+                 DemandMessage<K, V> msg = poll(queue, this);
+ 
+                 if (msg == null)
+                     continue;
+ 
+                 ClusterNode node = cctx.discovery().node(msg.senderId());
+ 
+                 if (node == null) {
+                     if (log.isDebugEnabled())
+                         log.debug("Received message from non-existing node (will ignore): " + msg);
+ 
+                     continue;
+                 }
+ 
+                 processMessage(msg, node);
+             }
+         }
+ 
+         /**
+          * @param msg Message.
+          * @param node Demander.
+          */
+         private void processMessage(DemandMessage<K, V> msg, ClusterNode node) {
+             assert msg != null;
+             assert node != null;
+ 
+             GridDhtPartitionDemandMessage<K, V> d = msg.message();
+ 
+             GridDhtPartitionSupplyMessage<K, V> s = new GridDhtPartitionSupplyMessage<>(d.workerId(),
+                 d.updateSequence(), cctx.cacheId());
+ 
+             long preloadThrottle = cctx.config().getPreloadThrottle();
+ 
+             boolean ack = false;
+ 
+             // If demander node left grid.
+             boolean nodeLeft = false;
+ 
+             boolean convertPortable = cctx.portableEnabled() && cctx.offheapTiered();
+ 
+             try {
+                 // Partition map exchange is finished which means that all near transactions with given
+                 // topology version are committed. We can wait for local locks here as it will not take
+                 // much time.
+                 cctx.mvcc().finishLocks(d.topologyVersion()).get();
+ 
+                 for (Integer part : d.partitions()) {
+                     GridDhtLocalPartition<K, V> loc = top.localPartition(part, d.topologyVersion(), false);
+ 
+                     if (loc == null || loc.state() != OWNING || !loc.reserve()) {
+                         // Reply with partition of "-1" to let sender know that
+                         // this node is no longer an owner.
+                         s.missed(part);
+ 
+                         if (log.isDebugEnabled())
+                             log.debug("Requested partition is not owned by local node [part=" + part +
+                                 ", demander=" + msg.senderId() + ']');
+ 
+                         continue;
+                     }
+ 
+                     GridCacheEntryInfoCollectSwapListener<K, V> swapLsnr = null;
+ 
+                     try {
+                         if (cctx.isSwapOrOffheapEnabled()) {
+                             swapLsnr = new GridCacheEntryInfoCollectSwapListener<>(log, cctx);
+ 
+                             cctx.swap().addOffHeapListener(part, swapLsnr);
+                             cctx.swap().addSwapListener(part, swapLsnr);
+                         }
+ 
+                         boolean partMissing = false;
+ 
+                         for (GridCacheEntryEx<K, V> e : loc.entries()) {
+                             if (!cctx.affinity().belongs(node, part, d.topologyVersion())) {
+                                 // Demander no longer needs this partition, so we send '-1' partition and move on.
+                                 s.missed(part);
+ 
+                                 if (log.isDebugEnabled())
+                                     log.debug("Demanding node does not need requested partition [part=" + part +
+                                         ", nodeId=" + msg.senderId() + ']');
+ 
+                                 partMissing = true;
+ 
+                                 break;
+                             }
+ 
+                             if (s.messageSize() >= cctx.config().getPreloadBatchSize()) {
+                                 ack = true;
+ 
+                                 if (!reply(node, d, s)) {
+                                     nodeLeft = true;
+ 
+                                     return;
+                                 }
+ 
+                                 // Throttle preloading.
+                                 if (preloadThrottle > 0)
+                                     U.sleep(preloadThrottle);
+ 
+                                 s = new GridDhtPartitionSupplyMessage<>(d.workerId(), d.updateSequence(),
+                                     cctx.cacheId());
+                             }
+ 
+                             GridCacheEntryInfo<K, V> info = e.info();
+ 
+                             if (info != null && !(info.key() instanceof GridPartitionLockKey) && !info.isNew()) {
+                                 if (preloadPred == null || preloadPred.apply(info))
+                                     s.addEntry(part, info, cctx.shared());
+                                 else if (log.isDebugEnabled())
+                                     log.debug("Preload predicate evaluated to false (will not sender cache entry): " +
+                                         info);
+                             }
+                         }
+ 
+                         if (partMissing)
+                             continue;
+ 
+                         if (cctx.isSwapOrOffheapEnabled()) {
+                             GridCloseableIterator<Map.Entry<byte[], GridCacheSwapEntry<V>>> iter =
+                                 cctx.swap().iterator(part, false);
+ 
+                             // Iterator may be null if space does not exist.
+                             if (iter != null) {
+                                 try {
+                                     boolean prepared = false;
+ 
+                                     for (Map.Entry<byte[], GridCacheSwapEntry<V>> e : iter) {
+                                         if (!cctx.affinity().belongs(node, part, d.topologyVersion())) {
+                                             // Demander no longer needs this partition,
+                                             // so we send '-1' partition and move on.
+                                             s.missed(part);
+ 
+                                             if (log.isDebugEnabled())
+                                                 log.debug("Demanding node does not need requested partition " +
+                                                     "[part=" + part + ", nodeId=" + msg.senderId() + ']');
+ 
+                                             partMissing = true;
+ 
+                                             break; // For.
+                                         }
+ 
+                                         if (s.messageSize() >= cctx.config().getPreloadBatchSize()) {
+                                             ack = true;
+ 
+                                             if (!reply(node, d, s)) {
+                                                 nodeLeft = true;
+ 
+                                                 return;
+                                             }
+ 
+                                             // Throttle preloading.
+                                             if (preloadThrottle > 0)
+                                                 U.sleep(preloadThrottle);
+ 
+                                             s = new GridDhtPartitionSupplyMessage<>(d.workerId(),
+                                                 d.updateSequence(), cctx.cacheId());
+                                         }
+ 
+                                         GridCacheSwapEntry<V> swapEntry = e.getValue();
+ 
+                                         GridCacheEntryInfo<K, V> info = new GridCacheEntryInfo<>();
+ 
+                                         info.keyBytes(e.getKey());
+                                         info.ttl(swapEntry.ttl());
+                                         info.expireTime(swapEntry.expireTime());
+                                         info.version(swapEntry.version());
+ 
+                                         if (!swapEntry.valueIsByteArray()) {
+                                             if (convertPortable)
+                                                 info.valueBytes(cctx.convertPortableBytes(swapEntry.valueBytes()));
+                                             else
+                                                 info.valueBytes(swapEntry.valueBytes());
+                                         }
+                                         else
+                                             info.value(swapEntry.value());
+ 
+                                         if (preloadPred == null || preloadPred.apply(info))
+                                             s.addEntry0(part, info, cctx.shared());
+                                         else {
+                                             if (log.isDebugEnabled())
+                                                 log.debug("Preload predicate evaluated to false (will not send " +
+                                                     "cache entry): " + info);
+ 
+                                             continue;
+                                         }
+ 
+                                         // Need to manually prepare cache message.
+                                         if (depEnabled && !prepared) {
+                                             ClassLoader ldr = swapEntry.keyClassLoaderId() != null ?
+                                                 cctx.deploy().getClassLoader(swapEntry.keyClassLoaderId()) :
+                                                 swapEntry.valueClassLoaderId() != null ?
+                                                     cctx.deploy().getClassLoader(swapEntry.valueClassLoaderId()) :
+                                                     null;
+ 
+                                             if (ldr == null)
+                                                 continue;
+ 
+                                             if (ldr instanceof GridDeploymentInfo) {
+                                                 s.prepare((GridDeploymentInfo)ldr);
+ 
+                                                 prepared = true;
+                                             }
+                                         }
+                                     }
+ 
+                                     if (partMissing)
+                                         continue;
+                                 }
+                                 finally {
+                                     iter.close();
+                                 }
+                             }
+                         }
+ 
+                         // Stop receiving promote notifications.
+                         if (swapLsnr != null) {
+                             cctx.swap().removeOffHeapListener(part, swapLsnr);
+                             cctx.swap().removeSwapListener(part, swapLsnr);
+                         }
+ 
+                         if (swapLsnr != null) {
+                             Collection<GridCacheEntryInfo<K, V>> entries = swapLsnr.entries();
+ 
+                             swapLsnr = null;
+ 
+                             for (GridCacheEntryInfo<K, V> info : entries) {
+                                 if (!cctx.affinity().belongs(node, part, d.topologyVersion())) {
+                                     // Demander no longer needs this partition,
+                                     // so we send '-1' partition and move on.
+                                     s.missed(part);
+ 
+                                     if (log.isDebugEnabled())
+                                         log.debug("Demanding node does not need requested partition " +
+                                             "[part=" + part + ", nodeId=" + msg.senderId() + ']');
+ 
+                                     // No need to continue iteration over swap entries.
+                                     break;
+                                 }
+ 
+                                 if (s.messageSize() >= cctx.config().getPreloadBatchSize()) {
+                                     ack = true;
+ 
+                                     if (!reply(node, d, s)) {
+                                         nodeLeft = true;
+ 
+                                         return;
+                                     }
+ 
+                                     s = new GridDhtPartitionSupplyMessage<>(d.workerId(), d.updateSequence(),
+                                         cctx.cacheId());
+                                 }
+ 
+                                 if (preloadPred == null || preloadPred.apply(info))
+                                     s.addEntry(part, info, cctx.shared());
+                                 else if (log.isDebugEnabled())
+                                     log.debug("Preload predicate evaluated to false (will not sender cache entry): " +
+                                         info);
+                             }
+                         }
+ 
+                         // Mark as last supply message.
+                         s.last(part);
+ 
+                         if (ack) {
+                             s.markAck();
+ 
+                             break; // Partition for loop.
+                         }
+                     }
+                     finally {
+                         loc.release();
+ 
+                         if (swapLsnr != null) {
+                             cctx.swap().removeOffHeapListener(part, swapLsnr);
+                             cctx.swap().removeSwapListener(part, swapLsnr);
+                         }
+                     }
+                 }
+ 
+                 reply(node, d, s);
+             }
+             catch (IgniteCheckedException e) {
+                 U.error(log, "Failed to send partition supply message to node: " + node.id(), e);
 -
 -                // Removing current topic because of request must fail with timeout and
 -                // demander will generate new topic.
 -                cctx.io().removeMessageId(d.topic());
 -            }
 -            finally {
 -                if (!ack || nodeLeft)
 -                    cctx.io().removeMessageId(d.topic());
+             }
+         }
+ 
+         /**
+          * @param n Node.
+          * @param d Demand message.
+          * @param s Supply message.
+          * @return {@code True} if message was sent, {@code false} if recipient left grid.
+          * @throws IgniteCheckedException If failed.
+          */
+         private boolean reply(ClusterNode n, GridDhtPartitionDemandMessage<K, V> d, GridDhtPartitionSupplyMessage<K, V> s)
+             throws IgniteCheckedException {
+             try {
+                 if (log.isDebugEnabled())
+                     log.debug("Replying to partition demand [node=" + n.id() + ", demand=" + d + ", supply=" + s + ']');
+ 
 -                cctx.io().sendOrderedMessage(n, d.topic(), cctx.io().messageId(d.topic(), n.id()), s, d.timeout());
++                cctx.io().sendOrderedMessage(n, d.topic(), s, d.timeout());
+ 
+                 return true;
+             }
+             catch (ClusterTopologyException ignore) {
+                 if (log.isDebugEnabled())
+                     log.debug("Failed to send partition supply message because node left grid: " + n.id());
+ 
+                 return false;
+             }
+         }
+     }
+ 
+     /**
+      * Demand message wrapper.
+      */
+     private static class DemandMessage<K, V> extends IgniteBiTuple<UUID, GridDhtPartitionDemandMessage<K, V>> {
+         /** */
+         private static final long serialVersionUID = 0L;
+ 
+         /**
+          * @param sndId Sender ID.
+          * @param msg Message.
+          */
+         DemandMessage(UUID sndId, GridDhtPartitionDemandMessage<K, V> msg) {
+             super(sndId, msg);
+         }
+ 
+         /**
+          * Empty constructor required for {@link Externalizable}.
+          */
+         public DemandMessage() {
+             // No-op.
+         }
+ 
+         /**
+          * @return Sender ID.
+          */
+         UUID senderId() {
+             return get1();
+         }
+ 
+         /**
+          * @return Message.
+          */
+         public GridDhtPartitionDemandMessage<K, V> message() {
+             return get2();
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return "DemandMessage [senderId=" + senderId() + ", msg=" + message() + ']';
+         }
+     }
+ }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
index 0000000,e7cd7f3..5be7d9f
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
@@@ -1,0 -1,793 +1,770 @@@
+ /*
+  * 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.query;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cache.*;
+ import org.apache.ignite.cache.query.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.events.*;
+ import org.apache.ignite.internal.util.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.internal.managers.eventstorage.*;
+ import org.apache.ignite.internal.processors.query.*;
+ 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.util.*;
+ import java.util.concurrent.*;
+ 
+ import static org.apache.ignite.events.IgniteEventType.*;
+ import static org.apache.ignite.cache.CacheMode.*;
+ import static org.apache.ignite.internal.GridTopic.*;
+ 
+ /**
+  * Distributed query manager.
+  */
+ public class GridCacheDistributedQueryManager<K, V> extends GridCacheQueryManager<K, V> {
+     /** */
+     private static final int MAX_CANCEL_IDS = 1000;
+ 
+     /** Query response frequency. */
+     private static final long RESEND_FREQ = 3000;
+ 
+     /** Query response attempts. */
+     private static final int RESEND_ATTEMPTS = 5;
+ 
+     /** Prefix for communication topic. */
+     private static final String TOPIC_PREFIX = "QUERY";
+ 
+     /** {request ID -> thread} */
+     private ConcurrentMap<Long, Thread> threads = new ConcurrentHashMap8<>();
+ 
+     /** {request ID -> future} */
+     private ConcurrentMap<Long, GridCacheDistributedQueryFuture<?, ?, ?>> futs =
+         new ConcurrentHashMap8<>();
+ 
+     /** Received requests to cancel. */
+     private Collection<CancelMessageId> cancelIds =
+         new GridBoundedConcurrentOrderedSet<>(MAX_CANCEL_IDS);
+ 
+     /** Canceled queries. */
+     private Collection<Long> cancelled = new GridBoundedConcurrentOrderedSet<>(MAX_CANCEL_IDS);
+ 
+     /** Query response handler. */
+     private IgniteBiInClosure<UUID,GridCacheQueryResponse<K,V>> resHnd = new CI2<UUID, GridCacheQueryResponse<K, V>>() {
+         @Override public void apply(UUID nodeId, GridCacheQueryResponse<K, V> res) {
+             processQueryResponse(nodeId, res);
+         }
+     };
+ 
+     /** {@inheritDoc} */
+     @Override public void start0() throws IgniteCheckedException {
+         super.start0();
+ 
+         assert cctx.config().getCacheMode() != LOCAL;
+ 
+         cctx.io().addHandler(cctx.cacheId(), GridCacheQueryRequest.class, new CI2<UUID, GridCacheQueryRequest<K, V>>() {
+             @Override public void apply(UUID nodeId, GridCacheQueryRequest<K, V> req) {
+                 processQueryRequest(nodeId, req);
+             }
+         });
+ 
+         cctx.events().addListener(new GridLocalEventListener() {
+             @Override public void onEvent(IgniteEvent evt) {
+                 IgniteDiscoveryEvent discoEvt = (IgniteDiscoveryEvent)evt;
+ 
+                 for (GridCacheDistributedQueryFuture fut : futs.values())
+                     fut.onNodeLeft(discoEvt.eventNode().id());
+             }
+         }, EVT_NODE_LEFT, EVT_NODE_FAILED);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void printMemoryStats() {
+         super.printMemoryStats();
+ 
+         X.println(">>>   threadsSize: " + threads.size());
+         X.println(">>>   futsSize: " + futs.size());
+     }
+ 
+     /**
+      * Removes query future from futures map.
+      *
+      * @param reqId Request id.
+      * @param fut Query future.
+      */
+     protected void addQueryFuture(long reqId, GridCacheDistributedQueryFuture<?, ?, ?> fut) {
+         futs.put(reqId, fut);
+     }
+ 
+     /**
+      * Removes query future from futures map.
+      *
+      * @param reqId Request id.
+      */
+     protected void removeQueryFuture(long reqId) {
+         futs.remove(reqId);
+     }
+ 
+     /**
+      * Gets query future from futures map.
+      *
+      * @param reqId Request id.
+      * @return Found future or null.
+      */
+     protected GridCacheDistributedQueryFuture<?, ?, ?> getQueryFuture(long reqId) {
+         return futs.get(reqId);
+     }
+ 
+     /**
+      * Processes cache query request.
+      *
+      * @param sndId Sender node id.
+      * @param req Query request.
+      */
+     @SuppressWarnings("unchecked")
+     @Override void processQueryRequest(UUID sndId, GridCacheQueryRequest req) {
+         if (req.cancel()) {
+             cancelIds.add(new CancelMessageId(req.id(), sndId));
+ 
+             if (req.fields())
+                 removeFieldsQueryResult(sndId, req.id());
+             else
+                 removeQueryResult(sndId, req.id());
+         }
+         else {
+             if (!cancelIds.contains(new CancelMessageId(req.id(), sndId))) {
+                 if (!F.eq(req.cacheName(), cctx.name())) {
+                     GridCacheQueryResponse res = new GridCacheQueryResponse(
+                         cctx.cacheId(),
+                         req.id(),
+                         new IgniteCheckedException("Received request for incorrect cache [expected=" + cctx.name() +
+                             ", actual=" + req.cacheName()));
+ 
+                     sendQueryResponse(sndId, res, 0);
+                 }
+                 else {
+                     threads.put(req.id(), Thread.currentThread());
+ 
+                     try {
+                         GridCacheQueryInfo info = distributedQueryInfo(sndId, req);
+ 
+                         if (info == null)
+                             return;
+ 
+                         if (req.fields())
+                             runFieldsQuery(info);
+                         else
+                             runQuery(info);
+                     }
+                     catch (Throwable e) {
+                         U.error(log(), "Failed to run query.", e);
+ 
+                         sendQueryResponse(sndId, new GridCacheQueryResponse(cctx.cacheId(), req.id(), e.getCause()), 0);
+                     }
+                     finally {
+                         threads.remove(req.id());
+                     }
+                 }
+             }
+         }
+     }
+ 
+     /**
+      * @param sndId Sender node id.
+      * @param req Query request.
+      * @return Query info.
+      * @throws ClassNotFoundException If class not found.
+      */
+     @Nullable private GridCacheQueryInfo distributedQueryInfo(UUID sndId, GridCacheQueryRequest<K, V> req)
+         throws ClassNotFoundException {
+         IgnitePredicate<CacheEntry<Object, Object>> prjPred = req.projectionFilter() == null ?
+             F.<CacheEntry<Object, Object>>alwaysTrue() : req.projectionFilter();
+ 
+         IgniteReducer<Object, Object> rdc = req.reducer();
+         IgniteClosure<Object, Object> trans = req.transformer();
+ 
+         ClusterNode sndNode = cctx.node(sndId);
+ 
+         if (sndNode == null)
+             return null;
+ 
+         GridCacheQueryAdapter<?> qry =
+             new GridCacheQueryAdapter<>(
+                 cctx,
+                 prjPred,
+                 req.type(),
+                 log,
+                 req.pageSize(),
+                 0,
+                 false,
+                 req.includeBackups(),
+                 false,
+                 null,
+                 req.keyValueFilter(),
+                 req.className(),
+                 req.clause(),
+                 req.includeMetaData(),
+                 req.keepPortable(),
+                 req.subjectId(),
+                 req.taskHash()
+             );
+ 
+         return new GridCacheQueryInfo(
+             false,
+             prjPred,
+             trans,
+             rdc,
+             qry,
+             null,
+             sndId,
+             req.id(),
+             req.includeMetaData(),
+             req.allPages(),
+             req.arguments()
+         );
+     }
+ 
+     /**
+      * Sends cache query response.
+      *
+      * @param nodeId Node to send response.
+      * @param res Cache query response.
+      * @param timeout Message timeout.
+      * @return {@code true} if response was sent, {@code false} otherwise.
+      */
+     private boolean sendQueryResponse(UUID nodeId, GridCacheQueryResponse<K, V> res, long timeout) {
+         ClusterNode node = cctx.node(nodeId);
+ 
+         if (node == null)
+             return false;
+ 
+         int attempt = 1;
+ 
+         IgniteCheckedException err = null;
+ 
+         while (!Thread.currentThread().isInterrupted()) {
+             try {
+                 if (log.isDebugEnabled())
+                     log.debug("Send query response: " + res);
+ 
+                 Object topic = topic(nodeId, res.requestId());
+ 
+                 cctx.io().sendOrderedMessage(
+                     node,
+                     topic,
 -                    cctx.io().messageId(topic, nodeId),
+                     res,
+                     timeout > 0 ? timeout : Long.MAX_VALUE);
+ 
+                 return true;
+             }
+             catch (ClusterTopologyException ignored) {
+                 if (log.isDebugEnabled())
+                     log.debug("Failed to send query response since node left grid [nodeId=" + nodeId +
+                         ", res=" + res + "]");
+ 
+                 return false;
+             }
+             catch (IgniteCheckedException e) {
+                 if (err == null)
+                     err = e;
+ 
+                 if (Thread.currentThread().isInterrupted())
+                     break;
+ 
+                 if (attempt < RESEND_ATTEMPTS) {
+                     if (log.isDebugEnabled())
+                         log.debug("Failed to send queries response (will try again) [nodeId=" + nodeId + ", res=" +
+                             res + ", attempt=" + attempt + ", err=" + e + "]");
+ 
+                     if (!Thread.currentThread().isInterrupted())
+                         try {
+                             U.sleep(RESEND_FREQ);
+                         }
+                         catch (IgniteInterruptedException e1) {
+                             U.error(log,
+                                 "Waiting for queries response resending was interrupted (response will not be sent) " +
+                                 "[nodeId=" + nodeId + ", response=" + res + "]", e1);
+ 
+                             return false;
+                         }
+                 }
+                 else {
+                     U.error(log, "Failed to sender cache response [nodeId=" + nodeId + ", response=" + res + "]", err);
+ 
+                     return false;
+                 }
+             }
+ 
+             attempt++;
+         }
+ 
+         return false;
+     }
+ 
 -    /** {@inheritDoc} */
 -    @Override protected void removeQueryResult(@Nullable UUID sndId, long reqId) {
 -        super.removeQueryResult(sndId, reqId);
 -
 -        if (sndId != null) {
 -            Object topic = topic(sndId, reqId);
 -
 -            cctx.io().removeMessageId(topic);
 -        }
 -    }
 -
 -    /** {@inheritDoc} */
 -    @Override protected void removeFieldsQueryResult(@Nullable UUID sndId, long reqId) {
 -        super.removeFieldsQueryResult(sndId, reqId);
 -
 -        if (sndId != null) {
 -            Object topic = topic(sndId, reqId);
 -
 -            cctx.io().removeMessageId(topic);
 -        }
 -    }
 -
+     /**
+      * Processes cache query response.
+      *
+      * @param sndId Sender node id.
+      * @param res Query response.
+      */
+     @SuppressWarnings("unchecked")
+     private void processQueryResponse(UUID sndId, GridCacheQueryResponse res) {
+         if (log.isDebugEnabled())
+             log.debug("Received query response: " + res);
+ 
+         GridCacheQueryFutureAdapter fut = getQueryFuture(res.requestId());
+ 
+         if (fut != null)
+             if (res.fields())
+                 ((GridCacheDistributedFieldsQueryFuture)fut).onPage(
+                     sndId, res.metadata(), res.data(), res.error(), res.isFinished());
+             else
+                 fut.onPage(sndId, res.data(), res.error(), res.isFinished());
+         else if (!cancelled.contains(res.requestId()))
+             U.warn(log, "Received response for finished or unknown query [rmtNodeId=" + sndId +
+                 ", res=" + res + ']');
+     }
+ 
+     /** {@inheritDoc} */
+     @Override void onQueryFutureCanceled(long reqId) {
+         cancelled.add(reqId);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override void onCancelAtStop() {
+         super.onCancelAtStop();
+ 
+         for (GridCacheQueryFutureAdapter fut : futs.values())
+             try {
+                 fut.cancel();
+             }
+             catch (IgniteCheckedException e) {
+                 U.error(log, "Failed to cancel running query future: " + fut, e);
+             }
+ 
+         U.interrupt(threads.values());
+     }
+ 
+     /** {@inheritDoc} */
+     @Override void onWaitAtStop() {
+         super.onWaitAtStop();
+ 
+         // Wait till all requests will be finished.
+         for (GridCacheQueryFutureAdapter fut : futs.values())
+             try {
+                 fut.get();
+             }
+             catch (IgniteCheckedException e) {
+                 if (log.isDebugEnabled())
+                     log.debug("Received query error while waiting for query to finish [queryFuture= " + fut +
+                         ", error= " + e + ']');
+             }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected boolean onPageReady(boolean loc, GridCacheQueryInfo qryInfo,
+         Collection<?> data, boolean finished, Throwable e) {
+         GridCacheLocalQueryFuture<?, ?, ?> fut = qryInfo.localQueryFuture();
+ 
+         if (loc)
+             assert fut != null;
+ 
+         if (e != null) {
+             if (loc)
+                 fut.onPage(null, null, e, true);
+             else
+                 sendQueryResponse(qryInfo.senderId(),
+                     new GridCacheQueryResponse<K, V>(cctx.cacheId(), qryInfo.requestId(), e),
+                     qryInfo.query().timeout());
+ 
+             return true;
+         }
+ 
+         if (loc)
+             fut.onPage(null, data, null, finished);
+         else {
+             GridCacheQueryResponse<K, V> res = new GridCacheQueryResponse<>(cctx.cacheId(), qryInfo.requestId(),
+                 /*finished*/false, /*fields*/false);
+ 
+             res.data(data);
+             res.finished(finished);
+ 
+             if (!sendQueryResponse(qryInfo.senderId(), res, qryInfo.query().timeout()))
+                 return false;
+         }
+ 
+         return true;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override protected boolean onFieldsPageReady(boolean loc, GridCacheQueryInfo qryInfo,
+         @Nullable List<GridQueryFieldMetadata> metadata,
+         @Nullable Collection<?> entities,
+         @Nullable Collection<?> data,
+         boolean finished, @Nullable Throwable e) {
+         assert qryInfo != null;
+ 
+         if (e != null) {
+             if (loc) {
+                 GridCacheLocalFieldsQueryFuture fut = (GridCacheLocalFieldsQueryFuture)qryInfo.localQueryFuture();
+ 
+                 fut.onPage(null, null, null, e, true);
+             }
+             else
+                 sendQueryResponse(qryInfo.senderId(),
+                     new GridCacheQueryResponse<K, V>(cctx.cacheId(), qryInfo.requestId(), e),
+                     qryInfo.query().timeout());
+ 
+             return true;
+         }
+ 
+         if (loc) {
+             GridCacheLocalFieldsQueryFuture fut = (GridCacheLocalFieldsQueryFuture)qryInfo.localQueryFuture();
+ 
+             fut.onPage(null, metadata, data, null, finished);
+         }
+         else {
+             GridCacheQueryResponse<K, V> res = new GridCacheQueryResponse<>(cctx.cacheId(), qryInfo.requestId(),
+                 finished, qryInfo.reducer() == null);
+ 
+             res.metadata(metadata);
+             res.data(entities != null ? entities : data);
+ 
+             if (!sendQueryResponse(qryInfo.senderId(), res, qryInfo.query().timeout()))
+                 return false;
+         }
+ 
+         return true;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public CacheQueryFuture<?> queryLocal(GridCacheQueryBean qry) {
+         assert cctx.config().getCacheMode() != LOCAL;
+ 
+         if (log.isDebugEnabled())
+             log.debug("Executing query on local node: " + qry);
+ 
+         GridCacheLocalQueryFuture<K, V, ?> fut = new GridCacheLocalQueryFuture<>(cctx, qry);
+ 
+         try {
+             qry.query().validate();
+ 
+             fut.execute();
+         }
+         catch (IgniteCheckedException e) {
+             fut.onDone(e);
+         }
+ 
+         return fut;
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public CacheQueryFuture<?> queryDistributed(GridCacheQueryBean qry, Collection<ClusterNode> nodes) {
+         assert cctx.config().getCacheMode() != LOCAL;
+ 
+         if (log.isDebugEnabled())
+             log.debug("Executing distributed query: " + qry);
+ 
+         long reqId = cctx.io().nextIoId();
+ 
+         final GridCacheDistributedQueryFuture<K, V, ?> fut =
+             new GridCacheDistributedQueryFuture<>(cctx, reqId, qry, nodes);
+ 
+         try {
+             qry.query().validate();
+ 
+             String clsName = qry.query().queryClassName();
+ 
+             GridCacheQueryRequest req = new GridCacheQueryRequest(
+                 cctx.cacheId(),
+                 reqId,
+                 cctx.name(),
+                 qry.query().type(),
+                 false,
+                 qry.query().clause(),
+                 clsName,
+                 qry.query().scanFilter(),
+                 qry.query().projectionFilter(),
+                 qry.reducer(),
+                 qry.transform(),
+                 qry.query().pageSize(),
+                 qry.query().includeBackups(),
+                 qry.arguments(),
+                 false,
+                 qry.query().keepPortable(),
+                 qry.query().subjectId(),
+                 qry.query().taskHash());
+ 
+             addQueryFuture(req.id(), fut);
+ 
+             final Object topic = topic(cctx.nodeId(), req.id());
+ 
+             cctx.io().addOrderedHandler(topic, resHnd);
+ 
+             fut.listenAsync(new CI1<IgniteFuture<?>>() {
+                 @Override public void apply(IgniteFuture<?> fut) {
+                     cctx.io().removeOrderedHandler(topic);
+                 }
+             });
+ 
+             sendRequest(fut, req, nodes);
+         }
+         catch (IgniteCheckedException e) {
+             fut.onDone(e);
+         }
+ 
+         return fut;
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void loadPage(long id, GridCacheQueryAdapter<?> qry, Collection<ClusterNode> nodes, boolean all) {
+         assert cctx.config().getCacheMode() != LOCAL;
+         assert qry != null;
+         assert nodes != null;
+ 
+         GridCacheDistributedQueryFuture<?, ?, ?> fut = futs.get(id);
+ 
+         assert fut != null;
+ 
+         try {
+             GridCacheQueryRequest<K, V> req = new GridCacheQueryRequest<>(
+                 cctx.cacheId(),
+                 id,
+                 cctx.name(),
+                 qry.pageSize(),
+                 qry.includeBackups(),
+                 fut.fields(),
+                 all,
+                 qry.keepPortable(),
+                 qry.subjectId(),
+                 qry.taskHash());
+ 
+             sendRequest(fut, req, nodes);
+         }
+         catch (IgniteCheckedException e) {
+             fut.onDone(e);
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public CacheQueryFuture<?> queryFieldsLocal(GridCacheQueryBean qry) {
+         assert cctx.config().getCacheMode() != LOCAL;
+ 
+         if (log.isDebugEnabled())
+             log.debug("Executing query on local node: " + qry);
+ 
+         GridCacheLocalFieldsQueryFuture fut = new GridCacheLocalFieldsQueryFuture(cctx, qry);
+ 
+         try {
+             qry.query().validate();
+ 
+             fut.execute();
+         }
+         catch (IgniteCheckedException e) {
+             fut.onDone(e);
+         }
+ 
+         return fut;
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("unchecked")
+     @Override public CacheQueryFuture<?> queryFieldsDistributed(GridCacheQueryBean qry,
+         Collection<ClusterNode> nodes) {
+         assert cctx.config().getCacheMode() != LOCAL;
+ 
+         if (log.isDebugEnabled())
+             log.debug("Executing distributed query: " + qry);
+ 
+         long reqId = cctx.io().nextIoId();
+ 
+         final GridCacheDistributedFieldsQueryFuture fut =
+             new GridCacheDistributedFieldsQueryFuture(cctx, reqId, qry, nodes);
+ 
+         try {
+             qry.query().validate();
+ 
+             GridCacheQueryRequest req = new GridCacheQueryRequest(
+                 cctx.cacheId(),
+                 reqId,
+                 cctx.name(),
+                 qry.query().type(),
+                 true,
+                 qry.query().clause(),
+                 null,
+                 null,
+                 qry.query().projectionFilter(),
+                 qry.reducer(),
+                 qry.transform(),
+                 qry.query().pageSize(),
+                 qry.query().includeBackups(),
+                 qry.arguments(),
+                 qry.query().includeMetadata(),
+                 qry.query().keepPortable(),
+                 qry.query().subjectId(),
+                 qry.query().taskHash());
+ 
+             addQueryFuture(req.id(), fut);
+ 
+             final Object topic = topic(cctx.nodeId(), req.id());
+ 
+             cctx.io().addOrderedHandler(topic, resHnd);
+ 
+             fut.listenAsync(new CI1<IgniteFuture<?>>() {
+                 @Override public void apply(IgniteFuture<?> fut) {
+                     cctx.io().removeOrderedHandler(topic);
+                 }
+             });
+ 
+             sendRequest(fut, req, nodes);
+         }
+         catch (IgniteCheckedException e) {
+             fut.onDone(e);
+         }
+ 
+         return fut;
+     }
+ 
+     /**
+      * Sends query request.
+      *
+      * @param fut Distributed future.
+      * @param req Request.
+      * @param nodes Nodes.
+      * @throws IgniteCheckedException In case of error.
+      * @deprecated Need to remove nodes filtration after breaking compatibility.
+      */
+     @Deprecated
+     @SuppressWarnings("unchecked")
+     private void sendRequest(
+         final GridCacheDistributedQueryFuture<?, ?, ?> fut,
+         final GridCacheQueryRequest<K, V> req,
+         Collection<ClusterNode> nodes
+     ) throws IgniteCheckedException {
+         assert fut != null;
+         assert req != null;
+         assert nodes != null;
+ 
+         final UUID locNodeId = cctx.localNodeId();
+ 
+         ClusterNode locNode = null;
+ 
+         Collection<ClusterNode> rmtNodes = null;
+ 
+         for (ClusterNode n : nodes) {
+             if (n.id().equals(locNodeId))
+                 locNode = n;
+             else {
+                 if (rmtNodes == null)
+                     rmtNodes = new ArrayList<>(nodes.size());
+ 
+                 rmtNodes.add(n);
+             }
+         }
+ 
+         // Request should be sent to remote nodes before the query is processed on the local node.
+         // For example, a remote reducer has a state, we should not serialize and then send
+         // the reducer changed by the local node.
+         if (!F.isEmpty(rmtNodes)) {
+             cctx.io().safeSend(rmtNodes, req, new P1<ClusterNode>() {
+                 @Override public boolean apply(ClusterNode node) {
+                     fut.onNodeLeft(node.id());
+ 
+                     return !fut.isDone();
+                 }
+             });
+         }
+ 
+         if (locNode != null) {
+             cctx.closures().callLocalSafe(new Callable<Object>() {
+                 @Override public Object call() throws Exception {
+                     req.beforeLocalExecution(cctx);
+ 
+                     processQueryRequest(locNodeId, req);
+ 
+                     return null;
+                 }
+             });
+         }
+     }
+ 
+     /**
+      * Gets topic for ordered response messages.
+      *
+      * @param nodeId Node ID.
+      * @param reqId Request ID.
+      * @return Topic.
+      */
+     private Object topic(UUID nodeId, long reqId) {
+         return TOPIC_CACHE.topic(TOPIC_PREFIX, nodeId, reqId);
+     }
+ 
+     /**
+      * Cancel message ID.
+      */
+     private static class CancelMessageId implements Comparable<CancelMessageId> {
+         /** Message ID. */
+         private long reqId;
+ 
+         /** Node ID. */
+         private UUID nodeId;
+ 
+         /**
+          * @param reqId Message ID.
+          * @param nodeId Node ID.
+          */
+         private CancelMessageId(long reqId, UUID nodeId) {
+             this.reqId = reqId;
+             this.nodeId = nodeId;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public int compareTo(CancelMessageId m) {
+             if (m.reqId == reqId)
+                 return m.nodeId.compareTo(nodeId);
+ 
+             return reqId < m.reqId ? -1 : 1;
+         }
+ 
+         /** {@inheritDoc} */
+         @SuppressWarnings({"unchecked"})
+         @Override public boolean equals(Object obj) {
+             if (obj == this)
+                 return true;
+ 
+             CancelMessageId other = (CancelMessageId)obj;
+ 
+             return reqId == other.reqId && nodeId.equals(other.nodeId);
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public int hashCode() {
+             return 31 * ((int)(reqId ^ (reqId >>> 32))) + nodeId.hashCode();
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return S.toString(CancelMessageId.class, this);
+         }
+     }
+ }


Mime
View raw message