| /* |
| * 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; |
| |
| import java.io.Externalizable; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.NoSuchElementException; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentMap; |
| import javax.cache.Cache; |
| import javax.cache.expiry.ExpiryPolicy; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.cache.ReadRepairStrategy; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.events.DiscoveryEvent; |
| import org.apache.ignite.events.Event; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.NodeStoppingException; |
| import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; |
| import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener; |
| import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; |
| import org.apache.ignite.internal.processors.cache.CacheObject; |
| import org.apache.ignite.internal.processors.cache.CacheOperationContext; |
| import org.apache.ignite.internal.processors.cache.EntryGetResult; |
| import org.apache.ignite.internal.processors.cache.EntryGetWithTtlResult; |
| import org.apache.ignite.internal.processors.cache.GridCacheAdapter; |
| import org.apache.ignite.internal.processors.cache.GridCacheClearAllRunnable; |
| import org.apache.ignite.internal.processors.cache.GridCacheConcurrentMap; |
| import org.apache.ignite.internal.processors.cache.GridCacheContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryEx; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException; |
| import org.apache.ignite.internal.processors.cache.GridCacheMapEntry; |
| import org.apache.ignite.internal.processors.cache.GridCachePreloader; |
| import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy; |
| import org.apache.ignite.internal.processors.cache.KeyCacheObject; |
| import org.apache.ignite.internal.processors.cache.ReaderArguments; |
| import org.apache.ignite.internal.processors.cache.distributed.GridCacheTtlUpdateRequest; |
| import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheAdapter; |
| import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheEntry; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtDetachedCacheEntry; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtForceKeysFuture; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtForceKeysRequest; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtForceKeysResponse; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology; |
| import org.apache.ignite.internal.processors.cache.distributed.near.CacheVersionedValue; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetResponse; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetResponse; |
| import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; |
| import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; |
| import org.apache.ignite.internal.processors.platform.cache.PlatformCacheEntryFilter; |
| import org.apache.ignite.internal.processors.security.SecurityUtils; |
| import org.apache.ignite.internal.util.future.GridEmbeddedFuture; |
| import org.apache.ignite.internal.util.future.GridFinishedFuture; |
| import org.apache.ignite.internal.util.future.GridFutureAdapter; |
| import org.apache.ignite.internal.util.lang.GridClosureException; |
| import org.apache.ignite.internal.util.lang.GridPlainRunnable; |
| import org.apache.ignite.internal.util.typedef.C2; |
| import org.apache.ignite.internal.util.typedef.CI1; |
| import org.apache.ignite.internal.util.typedef.CI2; |
| import org.apache.ignite.internal.util.typedef.CI3; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.internal.CU; |
| import org.apache.ignite.internal.util.typedef.internal.GPC; |
| import org.apache.ignite.internal.util.typedef.internal.S; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteBiInClosure; |
| import org.apache.ignite.lang.IgniteBiPredicate; |
| import org.apache.ignite.lang.IgniteBiTuple; |
| import org.apache.ignite.lang.IgniteUuid; |
| import org.apache.ignite.plugin.extensions.communication.Message; |
| import org.apache.ignite.plugin.security.SecurityPermission; |
| import org.jetbrains.annotations.NotNull; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.ignite.events.EventType.EVT_NODE_FAILED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_LEFT; |
| import static org.apache.ignite.internal.processors.dr.GridDrType.DR_LOAD; |
| import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE; |
| import static org.apache.ignite.internal.util.GridConcurrentFactory.newMap; |
| |
| /** |
| * DHT cache adapter. |
| */ |
| public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdapter<K, V> { |
| /** */ |
| private static final long serialVersionUID = 0L; |
| |
| /** Force key futures. */ |
| private final ConcurrentMap<IgniteUuid, GridDhtForceKeysFuture<?, ?>> forceKeyFuts = newMap(); |
| |
| /** */ |
| private volatile boolean stopping; |
| |
| /** Discovery listener. */ |
| private final GridLocalEventListener discoLsnr = new GridLocalEventListener() { |
| @Override public void onEvent(Event evt) { |
| DiscoveryEvent e = (DiscoveryEvent)evt; |
| |
| ClusterNode loc = ctx.localNode(); |
| |
| assert e.type() == EVT_NODE_LEFT || e.type() == EVT_NODE_FAILED : e; |
| |
| final ClusterNode n = e.eventNode(); |
| |
| assert !loc.id().equals(n.id()); |
| |
| for (GridDhtForceKeysFuture<?, ?> f : forceKeyFuts.values()) |
| f.onDiscoveryEvent(e); |
| } |
| }; |
| |
| /** |
| * Empty constructor required for {@link Externalizable}. |
| */ |
| protected GridDhtCacheAdapter() { |
| // No-op. |
| } |
| |
| /** |
| * Adds future to future map. |
| * |
| * @param fut Future to add. |
| * @return {@code False} if node cache is stopping and future was completed with error. |
| */ |
| public boolean addFuture(GridDhtForceKeysFuture<?, ?> fut) { |
| forceKeyFuts.put(fut.futureId(), fut); |
| |
| if (stopping) { |
| fut.onDone(stopError()); |
| |
| return false; |
| } |
| |
| return true; |
| } |
| |
| /** |
| * Removes future from future map. |
| * |
| * @param fut Future to remove. |
| */ |
| public void removeFuture(GridDhtForceKeysFuture<?, ?> fut) { |
| forceKeyFuts.remove(fut.futureId(), fut); |
| } |
| |
| /** |
| * @param node Node. |
| * @param msg Message. |
| */ |
| protected final void processForceKeyResponse(ClusterNode node, GridDhtForceKeysResponse msg) { |
| GridDhtForceKeysFuture<?, ?> f = forceKeyFuts.get(msg.futureId()); |
| |
| if (f != null) |
| f.onResult(msg); |
| else if (log.isDebugEnabled()) |
| log.debug("Receive force key response for unknown future (is it duplicate?) [nodeId=" + node.id() + |
| ", res=" + msg + ']'); |
| } |
| |
| /** |
| * @param node Node originated request. |
| * @param msg Force keys message. |
| */ |
| protected final void processForceKeysRequest(final ClusterNode node, final GridDhtForceKeysRequest msg) { |
| IgniteInternalFuture<?> fut = ctx.mvcc().finishKeys(msg.keys(), msg.cacheId(), msg.topologyVersion()); |
| |
| if (fut.isDone()) |
| processForceKeysRequest0(node, msg); |
| else |
| fut.listen(new CI1<IgniteInternalFuture<?>>() { |
| @Override public void apply(IgniteInternalFuture<?> t) { |
| processForceKeysRequest0(node, msg); |
| } |
| }); |
| } |
| |
| /** |
| * @param node Node originated request. |
| * @param msg Force keys message. |
| */ |
| private void processForceKeysRequest0(ClusterNode node, GridDhtForceKeysRequest msg) { |
| try { |
| ClusterNode loc = ctx.localNode(); |
| |
| GridDhtForceKeysResponse res = new GridDhtForceKeysResponse( |
| ctx.cacheId(), |
| msg.futureId(), |
| msg.miniId(), |
| ctx.deploymentEnabled()); |
| |
| GridDhtPartitionTopology top = ctx.topology(); |
| |
| for (KeyCacheObject k : msg.keys()) { |
| int p = ctx.affinity().partition(k); |
| |
| GridDhtLocalPartition locPart = top.localPartition(p, AffinityTopologyVersion.NONE, false); |
| |
| // If this node is no longer an owner. |
| if (locPart == null && !top.owners(p).contains(loc)) { |
| res.addMissed(k); |
| |
| continue; |
| } |
| |
| GridCacheEntryEx entry; |
| |
| while (true) { |
| ctx.shared().database().checkpointReadLock(); |
| |
| try { |
| entry = ctx.dht().entryEx(k); |
| |
| entry.unswap(); |
| |
| GridCacheEntryInfo info = entry.info(); |
| |
| if (info == null) { |
| assert entry.obsolete() : entry; |
| |
| continue; |
| } |
| |
| if (!info.isNew()) |
| res.addInfo(info); |
| |
| entry.touch(); |
| |
| break; |
| } |
| catch (GridCacheEntryRemovedException ignore) { |
| if (log.isDebugEnabled()) |
| log.debug("Got removed entry: " + k); |
| } |
| catch (GridDhtInvalidPartitionException ignore) { |
| if (log.isDebugEnabled()) |
| log.debug("Local node is no longer an owner: " + p); |
| |
| res.addMissed(k); |
| |
| break; |
| } |
| finally { |
| ctx.shared().database().checkpointReadUnlock(); |
| } |
| } |
| } |
| |
| if (log.isDebugEnabled()) |
| log.debug("Sending force key response [node=" + node.id() + ", res=" + res + ']'); |
| |
| ctx.io().send(node, res, ctx.ioPolicy()); |
| } |
| catch (ClusterTopologyCheckedException ignore) { |
| if (log.isDebugEnabled()) |
| log.debug("Received force key request form failed node (will ignore) [nodeId=" + node.id() + |
| ", req=" + msg + ']'); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to reply to force key request [nodeId=" + node.id() + ", req=" + msg + ']', e); |
| } |
| } |
| |
| /** |
| * |
| */ |
| public void dumpDebugInfo() { |
| if (!forceKeyFuts.isEmpty()) { |
| U.warn(log, "Pending force key futures [cache=" + ctx.name() + "]:"); |
| |
| for (GridDhtForceKeysFuture fut : forceKeyFuts.values()) |
| U.warn(log, ">>> " + fut); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onKernalStop() { |
| super.onKernalStop(); |
| |
| stopping = true; |
| |
| IgniteCheckedException err = stopError(); |
| |
| for (GridDhtForceKeysFuture fut : forceKeyFuts.values()) |
| fut.onDone(err); |
| |
| ctx.gridEvents().removeLocalEventListener(discoLsnr); |
| } |
| |
| /** |
| * @return Node stop exception. |
| */ |
| private IgniteCheckedException stopError() { |
| return new NodeStoppingException("Operation has been cancelled (cache or node is stopping)."); |
| } |
| |
| /** |
| * @param nodeId Sender node ID. |
| * @param res Near get response. |
| */ |
| protected final void processNearGetResponse(UUID nodeId, GridNearGetResponse res) { |
| if (log.isDebugEnabled()) |
| log.debug("Processing near get response [nodeId=" + nodeId + ", res=" + res + ']'); |
| |
| CacheGetFuture fut = (CacheGetFuture)ctx.mvcc().future(res.futureId()); |
| |
| if (fut == null) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to find future for get response [sender=" + nodeId + ", res=" + res + ']'); |
| |
| return; |
| } |
| |
| fut.onResult(nodeId, res); |
| } |
| |
| /** |
| * @param nodeId Sender node ID. |
| * @param res Near get response. |
| */ |
| protected void processNearSingleGetResponse(UUID nodeId, GridNearSingleGetResponse res) { |
| if (log.isDebugEnabled()) |
| log.debug("Processing near get response [nodeId=" + nodeId + ", res=" + res + ']'); |
| |
| GridPartitionedSingleGetFuture fut = (GridPartitionedSingleGetFuture)ctx.mvcc() |
| .future(new IgniteUuid(IgniteUuid.VM_ID, res.futureId())); |
| |
| if (fut == null) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to find future for get response [sender=" + nodeId + ", res=" + res + ']'); |
| |
| return; |
| } |
| |
| fut.onResult(nodeId, res); |
| } |
| |
| /** |
| * @param ctx Context. |
| */ |
| protected GridDhtCacheAdapter(GridCacheContext<K, V> ctx) { |
| this(ctx, new GridCachePartitionedConcurrentMap(ctx.group())); |
| } |
| |
| /** |
| * Constructor used for near-only cache. |
| * |
| * @param ctx Cache context. |
| * @param map Cache map. |
| */ |
| protected GridDhtCacheAdapter(GridCacheContext<K, V> ctx, GridCacheConcurrentMap map) { |
| super(ctx, map); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onKernalStart() throws IgniteCheckedException { |
| super.onKernalStart(); |
| |
| assert !ctx.isRecoveryMode() : "Registering message handlers in recovery mode [cacheName=" + name() + ']'; |
| |
| ctx.io().addCacheHandler(ctx.cacheId(), ctx.startTopologyVersion(), GridCacheTtlUpdateRequest.class, |
| (CI2<UUID, GridCacheTtlUpdateRequest>)this::processTtlUpdateRequest); |
| |
| ctx.gridEvents().addLocalEventListener(discoLsnr, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void printMemoryStats() { |
| super.printMemoryStats(); |
| |
| ctx.group().topology().printMemoryStats(1024); |
| } |
| |
| /** |
| * @return Near cache. |
| */ |
| public abstract GridNearCacheAdapter<K, V> near(); |
| |
| /** |
| * @return Partition topology. |
| */ |
| public GridDhtPartitionTopology topology() { |
| return ctx.group().topology(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public GridCachePreloader preloader() { |
| return ctx.group().preloader(); |
| } |
| |
| /** |
| * @param key Key. |
| * @return DHT entry. |
| */ |
| @Nullable public GridDhtCacheEntry peekExx(KeyCacheObject key) { |
| return (GridDhtCacheEntry)peekEx(key); |
| } |
| |
| /** |
| * {@inheritDoc} |
| * |
| * @throws GridDhtInvalidPartitionException If partition for the key is no longer valid. |
| */ |
| @Override public GridCacheEntryEx entryEx(KeyCacheObject key, |
| AffinityTopologyVersion topVer) throws GridDhtInvalidPartitionException { |
| return super.entryEx(key, topVer); |
| } |
| |
| /** |
| * @param key Key. |
| * @return DHT entry. |
| * @throws GridDhtInvalidPartitionException If partition for the key is no longer valid. |
| */ |
| public GridDhtCacheEntry entryExx(KeyCacheObject key) throws GridDhtInvalidPartitionException { |
| return (GridDhtCacheEntry)entryEx(key); |
| } |
| |
| /** |
| * @param key Key. |
| * @param topVer Topology version. |
| * @return DHT entry. |
| * @throws GridDhtInvalidPartitionException If partition for the key is no longer valid. |
| */ |
| public GridDhtCacheEntry entryExx(KeyCacheObject key, |
| AffinityTopologyVersion topVer) throws GridDhtInvalidPartitionException { |
| return (GridDhtCacheEntry)entryEx(key, topVer); |
| } |
| |
| /** |
| * @param key Key for which entry should be returned. |
| * @return Cache entry. |
| */ |
| protected GridDistributedCacheEntry createEntry(KeyCacheObject key) { |
| return new GridDhtDetachedCacheEntry(ctx, key); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void localLoad(Collection<? extends K> keys, final ExpiryPolicy plc, final boolean keepBinary) |
| throws IgniteCheckedException { |
| if (ctx.store().isLocal()) { |
| super.localLoad(keys, plc, keepBinary); |
| |
| return; |
| } |
| |
| // Version for all loaded entries. |
| final AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion(); |
| |
| final GridCacheVersion ver0 = ctx.shared().versions().nextForLoad(topVer.topologyVersion()); |
| |
| final boolean replicate = ctx.isDrEnabled(); |
| |
| final ExpiryPolicy plc0 = plc != null ? plc : ctx.expiry(); |
| |
| Collection<KeyCacheObject> keys0 = ctx.cacheKeysView(keys); |
| |
| ctx.store().loadAll(null, keys0, new CI2<KeyCacheObject, Object>() { |
| @Override public void apply(KeyCacheObject key, Object val) { |
| loadEntry(key, val, ver0, null, topVer, replicate, plc0); |
| } |
| }); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void localLoadCache(final IgniteBiPredicate<K, V> p, Object[] args) throws IgniteCheckedException { |
| if (ctx.store().isLocal()) { |
| super.localLoadCache(p, args); |
| |
| return; |
| } |
| |
| final AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion(); |
| |
| // Version for all loaded entries. |
| final GridCacheVersion ver0 = ctx.shared().versions().nextForLoad(topVer.topologyVersion()); |
| |
| final boolean replicate = ctx.isDrEnabled(); |
| |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| ExpiryPolicy plc0 = opCtx != null ? opCtx.expiry() : null; |
| |
| final ExpiryPolicy plc = plc0 != null ? plc0 : ctx.expiry(); |
| |
| final IgniteBiPredicate<K, V> pred; |
| |
| if (p != null) { |
| ctx.kernalContext().resource().injectGeneric(p); |
| |
| pred = SecurityUtils.sandboxedProxy(ctx.kernalContext(), IgniteBiPredicate.class, p); |
| } |
| else |
| pred = null; |
| |
| try { |
| ctx.store().loadCache(new CI3<KeyCacheObject, Object, GridCacheVersion>() { |
| @Override public void apply(KeyCacheObject key, Object val, @Nullable GridCacheVersion ver) { |
| assert ver == null; |
| |
| loadEntry(key, val, ver0, pred, topVer, replicate, plc); |
| } |
| }, args); |
| |
| } |
| finally { |
| if (p instanceof PlatformCacheEntryFilter) |
| ((PlatformCacheEntryFilter)p).onClose(); |
| } |
| } |
| |
| /** |
| * @param key Key. |
| * @param val Value. |
| * @param ver Cache version. |
| * @param p Optional predicate. |
| * @param topVer Topology version. |
| * @param replicate Replication flag. |
| * @param plc Expiry policy. |
| */ |
| private void loadEntry(KeyCacheObject key, |
| Object val, |
| GridCacheVersion ver, |
| @Nullable IgniteBiPredicate<K, V> p, |
| AffinityTopologyVersion topVer, |
| boolean replicate, |
| @Nullable ExpiryPolicy plc) { |
| if (p != null && !p.apply(key.<K>value(ctx.cacheObjectContext(), false), (V)val)) |
| return; |
| |
| try { |
| GridDhtLocalPartition part = ctx.group().topology().localPartition(ctx.affinity().partition(key), |
| AffinityTopologyVersion.NONE, true); |
| |
| // Reserve to make sure that partition does not get unloaded. |
| if (part.reserve()) { |
| GridCacheEntryEx entry = null; |
| |
| ctx.shared().database().checkpointReadLock(); |
| |
| try { |
| long ttl = CU.ttlForLoad(plc); |
| |
| if (ttl == CU.TTL_ZERO) |
| return; |
| |
| CacheObject cacheVal = ctx.toCacheObject(val); |
| |
| entry = entryEx(key); |
| |
| entry.initialValue(cacheVal, |
| ver, |
| ttl, |
| CU.EXPIRE_TIME_CALCULATE, |
| false, |
| topVer, |
| replicate ? DR_LOAD : DR_NONE, |
| true, |
| false); |
| } |
| catch (IgniteCheckedException e) { |
| throw new IgniteException("Failed to put cache value: " + entry, e); |
| } |
| catch (GridCacheEntryRemovedException ignore) { |
| if (log.isDebugEnabled()) |
| log.debug("Got removed entry during loadCache (will ignore): " + entry); |
| } |
| finally { |
| if (entry != null) |
| entry.touch(); |
| |
| part.release(); |
| |
| ctx.shared().database().checkpointReadUnlock(); |
| } |
| } |
| else if (log.isDebugEnabled()) |
| log.debug("Will node load entry into cache (partition is invalid): " + part); |
| } |
| catch (GridDhtInvalidPartitionException e) { |
| if (log.isDebugEnabled()) |
| log.debug(S.toString("Ignoring entry for partition that does not belong", |
| "key", key, true, |
| "val", val, true, |
| "err", e, false)); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public int size() { |
| return (int)sizeLong(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public long sizeLong() { |
| long sum = 0; |
| |
| for (GridDhtLocalPartition p : topology().currentLocalPartitions()) |
| sum += p.dataStore().cacheSize(ctx.cacheId()); |
| |
| return sum; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public int primarySize() { |
| return (int)primarySizeLong(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public long primarySizeLong() { |
| long sum = 0; |
| |
| AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion(); |
| |
| for (GridDhtLocalPartition p : topology().currentLocalPartitions()) { |
| if (p.primary(topVer)) |
| sum += p.dataStore().cacheSize(ctx.cacheId()); |
| } |
| |
| return sum; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<Map<K, V>> getAllAsync( |
| @Nullable Collection<? extends K> keys, |
| boolean forcePrimary, |
| boolean skipTx, |
| String taskName, |
| boolean deserializeBinary, |
| boolean recovery, |
| ReadRepairStrategy readRepairStrategy, |
| boolean skipVals, |
| boolean needVer |
| ) { |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| ctx.checkSecurity(SecurityPermission.CACHE_READ); |
| |
| warnIfUnordered(keys, BulkOperation.GET); |
| |
| return getAllAsync0( |
| ctx.cacheKeysView(keys), |
| null, |
| opCtx == null || !opCtx.skipStore(), |
| taskName, |
| deserializeBinary, |
| null, |
| skipVals, |
| /*keep cache objects*/false, |
| opCtx != null && opCtx.recovery(), |
| needVer, |
| null); |
| } |
| |
| /** |
| * @param keys Keys. |
| * @param readerArgs Near cache reader will be added if not null. |
| * @param readThrough Read-through flag. |
| * @param taskName Task name/ |
| * @param deserializeBinary Deserialize binary flag. |
| * @param expiry Expiry policy. |
| * @param skipVals Skip values flag. |
| * @param keepCacheObjects Keep cache objects. |
| * @param needVer If {@code true} returns values as tuples containing value and version. |
| * @param txLbl Transaction label. |
| * @return Future. |
| */ |
| private <K1, V1> IgniteInternalFuture<Map<K1, V1>> getAllAsync0( |
| @Nullable final Collection<KeyCacheObject> keys, |
| @Nullable final ReaderArguments readerArgs, |
| final boolean readThrough, |
| final String taskName, |
| final boolean deserializeBinary, |
| @Nullable final IgniteCacheExpiryPolicy expiry, |
| final boolean skipVals, |
| final boolean keepCacheObjects, |
| final boolean recovery, |
| final boolean needVer, |
| @Nullable String txLbl |
| ) { |
| if (F.isEmpty(keys)) |
| return new GridFinishedFuture<>(Collections.emptyMap()); |
| |
| Map<KeyCacheObject, EntryGetResult> misses = null; |
| |
| Set<GridCacheEntryEx> newLocEntries = null; |
| |
| ctx.shared().database().checkpointReadLock(); |
| |
| try { |
| int keysSize = keys.size(); |
| |
| GridDhtTopologyFuture topFut = ctx.shared().exchange().lastFinishedFuture(); |
| |
| Throwable ex = topFut != null ? topFut.validateCache(ctx, recovery, /*read*/true, null, keys) : null; |
| |
| if (ex != null) |
| return new GridFinishedFuture<>(ex); |
| |
| final Map<K1, V1> map = keysSize == 1 |
| ? (Map<K1, V1>)new IgniteBiTuple<>() |
| : U.newHashMap(keysSize); |
| |
| final boolean storeEnabled = !skipVals && readThrough && ctx.readThrough(); |
| |
| boolean readNoEntry = ctx.readNoEntry(expiry, readerArgs != null); |
| |
| for (KeyCacheObject key : keys) { |
| while (true) { |
| try { |
| EntryGetResult res = null; |
| |
| boolean evt = !skipVals; |
| boolean updateMetrics = !skipVals; |
| |
| GridCacheEntryEx entry = null; |
| |
| boolean skipEntry = readNoEntry; |
| |
| if (readNoEntry) { |
| CacheDataRow row = ctx.offheap().read(ctx, key); |
| |
| if (row != null) { |
| long expireTime = row.expireTime(); |
| |
| if (expireTime != 0) { |
| if (expireTime > U.currentTimeMillis()) { |
| res = new EntryGetWithTtlResult(row.value(), |
| row.version(), |
| false, |
| expireTime, |
| 0); |
| } |
| else |
| skipEntry = false; |
| } |
| else |
| res = new EntryGetResult(row.value(), row.version(), false); |
| } |
| |
| if (res != null) { |
| if (evt) { |
| ctx.events().readEvent(key, |
| null, |
| txLbl, |
| row.value(), |
| taskName, |
| !deserializeBinary); |
| } |
| |
| if (updateMetrics && ctx.statisticsEnabled()) |
| ctx.cache().metrics0().onRead(true); |
| } |
| else if (storeEnabled) |
| skipEntry = false; |
| } |
| |
| if (!skipEntry) { |
| boolean isNewLocEntry = this.map.getEntry(ctx, key) == null; |
| |
| entry = entryEx(key); |
| |
| if (entry == null) { |
| if (!skipVals && ctx.statisticsEnabled()) |
| ctx.cache().metrics0().onRead(false); |
| |
| break; |
| } |
| |
| if (isNewLocEntry) { |
| if (newLocEntries == null) |
| newLocEntries = new HashSet<>(); |
| |
| newLocEntries.add(entry); |
| } |
| |
| if (storeEnabled) { |
| res = entry.innerGetAndReserveForLoad(updateMetrics, |
| evt, |
| taskName, |
| expiry, |
| !deserializeBinary, |
| readerArgs); |
| |
| assert res != null; |
| |
| if (res.value() == null) { |
| if (misses == null) |
| misses = new HashMap<>(); |
| |
| misses.put(key, res); |
| |
| res = null; |
| } |
| } |
| else { |
| res = entry.innerGetVersioned( |
| null, |
| null, |
| updateMetrics, |
| evt, |
| null, |
| taskName, |
| expiry, |
| !deserializeBinary, |
| readerArgs); |
| |
| if (res == null) |
| entry.touch(); |
| } |
| } |
| |
| if (res != null) { |
| ctx.addResult(map, |
| key, |
| res, |
| skipVals, |
| keepCacheObjects, |
| deserializeBinary, |
| true, |
| needVer); |
| |
| if (entry != null) |
| entry.touch(); |
| |
| if (keysSize == 1) |
| // Safe to return because no locks are required in READ_COMMITTED mode. |
| return new GridFinishedFuture<>(map); |
| } |
| |
| break; |
| } |
| catch (GridCacheEntryRemovedException ignored) { |
| if (log.isDebugEnabled()) |
| log.debug("Got removed entry in getAllAsync(..) method (will retry): " + key); |
| } |
| } |
| } |
| |
| if (storeEnabled && misses != null) { |
| final Map<KeyCacheObject, EntryGetResult> loadKeys = misses; |
| |
| final Collection<KeyCacheObject> loaded = new HashSet<>(); |
| |
| return new GridEmbeddedFuture( |
| ctx.closures().callLocalSafe(ctx.projectSafe(new GPC<Map<K1, V1>>() { |
| @Override public Map<K1, V1> call() throws Exception { |
| ctx.store().loadAll(null/*tx*/, loadKeys.keySet(), new CI2<KeyCacheObject, Object>() { |
| @Override public void apply(KeyCacheObject key, Object val) { |
| EntryGetResult res = loadKeys.get(key); |
| |
| if (res == null || val == null) |
| return; |
| |
| loaded.add(key); |
| |
| CacheObject cacheVal = ctx.toCacheObject(val); |
| |
| while (true) { |
| GridCacheEntryEx entry = null; |
| |
| try { |
| ctx.shared().database().ensureFreeSpace(ctx.dataRegion()); |
| } |
| catch (IgniteCheckedException e) { |
| // Wrap errors (will be unwrapped). |
| throw new GridClosureException(e); |
| } |
| |
| ctx.shared().database().checkpointReadLock(); |
| |
| try { |
| entry = entryEx(key); |
| |
| entry.unswap(); |
| |
| GridCacheVersion newVer = nextVersion(); |
| |
| EntryGetResult verVal = entry.versionedValue( |
| cacheVal, |
| res.version(), |
| newVer, |
| expiry, |
| readerArgs); |
| |
| if (log.isDebugEnabled()) |
| log.debug("Set value loaded from store into entry [" + |
| "oldVer=" + res.version() + |
| ", newVer=" + verVal.version() + ", " + |
| "entry=" + entry + ']'); |
| |
| // Don't put key-value pair into result map if value is null. |
| if (verVal.value() != null) { |
| ctx.addResult(map, |
| key, |
| verVal, |
| skipVals, |
| keepCacheObjects, |
| deserializeBinary, |
| true, |
| needVer); |
| } |
| else { |
| ctx.addResult( |
| map, |
| key, |
| new EntryGetResult(cacheVal, res.version()), |
| skipVals, |
| keepCacheObjects, |
| deserializeBinary, |
| false, |
| needVer |
| ); |
| } |
| |
| entry.touch(); |
| |
| break; |
| } |
| catch (GridCacheEntryRemovedException ignore) { |
| if (log.isDebugEnabled()) |
| log.debug("Got removed entry during getAllAsync (will retry): " + |
| entry); |
| } |
| catch (IgniteCheckedException e) { |
| // Wrap errors (will be unwrapped). |
| throw new GridClosureException(e); |
| } |
| finally { |
| ctx.shared().database().checkpointReadUnlock(); |
| } |
| } |
| } |
| }); |
| |
| clearReservationsIfNeeded(loadKeys, loaded); |
| |
| return map; |
| } |
| }), true), |
| new C2<Map<K, V>, Exception, IgniteInternalFuture<Map<K, V>>>() { |
| @Override public IgniteInternalFuture<Map<K, V>> apply(Map<K, V> map, Exception e) { |
| if (e != null) { |
| clearReservationsIfNeeded(loadKeys, loaded); |
| |
| return new GridFinishedFuture<>(e); |
| } |
| |
| Collection<KeyCacheObject> notFound = new HashSet<>(loadKeys.keySet()); |
| |
| notFound.removeAll(loaded); |
| |
| // Touch entries that were not found in store. |
| for (KeyCacheObject key : notFound) { |
| GridCacheEntryEx entry = peekEx(key); |
| |
| if (entry != null) |
| entry.touch(); |
| } |
| |
| // There were no misses. |
| return new GridFinishedFuture<>(Collections.emptyMap()); |
| } |
| }, |
| new C2<Map<K1, V1>, Exception, Map<K1, V1>>() { |
| @Override public Map<K1, V1> apply(Map<K1, V1> loaded, Exception e) { |
| if (e == null) |
| map.putAll(loaded); |
| |
| return map; |
| } |
| } |
| ); |
| } |
| else |
| // Misses can be non-zero only if store is enabled. |
| assert misses == null; |
| |
| return new GridFinishedFuture<>(map); |
| } |
| catch (RuntimeException | AssertionError e) { |
| if (misses != null) { |
| for (KeyCacheObject key0 : misses.keySet()) { |
| GridCacheEntryEx entry = peekEx(key0); |
| if (entry != null) |
| entry.touch(); |
| } |
| } |
| |
| if (newLocEntries != null) { |
| for (GridCacheEntryEx entry : newLocEntries) |
| removeEntry(entry); |
| } |
| |
| return new GridFinishedFuture<>(e); |
| } |
| catch (IgniteCheckedException e) { |
| return new GridFinishedFuture<>(e); |
| } |
| finally { |
| ctx.shared().database().checkpointReadUnlock(); |
| } |
| } |
| |
| /** |
| * @param loadKeys Keys to load. |
| * @param loaded Actually loaded keys. |
| */ |
| private void clearReservationsIfNeeded(Map<KeyCacheObject, EntryGetResult> loadKeys, Collection<KeyCacheObject> loaded) { |
| if (loaded.size() != loadKeys.size()) { |
| for (Map.Entry<KeyCacheObject, EntryGetResult> e : loadKeys.entrySet()) { |
| if (loaded.contains(e.getKey())) |
| continue; |
| |
| GridCacheEntryEx entry = peekEx(e.getKey()); |
| |
| if (entry != null) { |
| if (e.getValue().reserved()) |
| entry.clearReserveForLoad(e.getValue().version()); |
| |
| entry.touch(); |
| } |
| } |
| } |
| } |
| |
| /** |
| * @param keys Keys to get |
| * @param readerArgs Reader will be added if not null. |
| * @param readThrough Read through flag. |
| * @param taskName Task name. |
| * @param expiry Expiry policy. |
| * @param skipVals Skip values flag. |
| * @param txLbl Transaction label. |
| * @return Get future. |
| */ |
| IgniteInternalFuture<Map<KeyCacheObject, EntryGetResult>> getDhtAllAsync( |
| Collection<KeyCacheObject> keys, |
| @Nullable final ReaderArguments readerArgs, |
| boolean readThrough, |
| String taskName, |
| @Nullable IgniteCacheExpiryPolicy expiry, |
| boolean skipVals, |
| boolean recovery, |
| @Nullable String txLbl |
| ) { |
| return getAllAsync0(keys, |
| readerArgs, |
| readThrough, |
| taskName, |
| false, |
| expiry, |
| skipVals, |
| /*keep cache objects*/true, |
| recovery, |
| /*need version*/true, |
| txLbl); |
| } |
| |
| /** |
| * @param reader Reader node ID. |
| * @param msgId Message ID. |
| * @param keys Keys to get. |
| * @param addReaders Add readers flag. |
| * @param readThrough Read through flag. |
| * @param topVer Topology version. |
| * @param taskNameHash Task name hash code. |
| * @param expiry Expiry policy. |
| * @param skipVals Skip values flag. |
| * @param txLbl Transaction label. |
| * @return DHT future. |
| */ |
| public GridDhtFuture<Collection<GridCacheEntryInfo>> getDhtAsync(UUID reader, |
| long msgId, |
| Map<KeyCacheObject, Boolean> keys, |
| boolean addReaders, |
| boolean readThrough, |
| AffinityTopologyVersion topVer, |
| int taskNameHash, |
| @Nullable IgniteCacheExpiryPolicy expiry, |
| boolean skipVals, |
| boolean recovery, |
| @Nullable String txLbl |
| ) { |
| GridDhtGetFuture<K, V> fut = new GridDhtGetFuture<>(ctx, |
| msgId, |
| reader, |
| keys, |
| readThrough, |
| topVer, |
| taskNameHash, |
| expiry, |
| skipVals, |
| recovery, |
| addReaders, |
| txLbl); |
| |
| fut.init(); |
| |
| return fut; |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param msgId Message ID. |
| * @param key Key. |
| * @param addRdr Add reader flag. |
| * @param readThrough Read through flag. |
| * @param topVer Topology version flag. |
| * @param taskNameHash Task name hash. |
| * @param expiry Expiry. |
| * @param skipVals Skip vals flag. |
| * @param txLbl Transaction label. |
| * @return Future for the operation. |
| */ |
| GridDhtGetSingleFuture getDhtSingleAsync( |
| UUID nodeId, |
| long msgId, |
| KeyCacheObject key, |
| boolean addRdr, |
| boolean readThrough, |
| AffinityTopologyVersion topVer, |
| int taskNameHash, |
| @Nullable IgniteCacheExpiryPolicy expiry, |
| boolean skipVals, |
| boolean recovery, |
| String txLbl |
| ) { |
| GridDhtGetSingleFuture fut = new GridDhtGetSingleFuture<>( |
| ctx, |
| msgId, |
| nodeId, |
| key, |
| addRdr, |
| readThrough, |
| topVer, |
| taskNameHash, |
| expiry, |
| skipVals, |
| recovery, |
| txLbl); |
| |
| fut.init(); |
| |
| return fut; |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param req Get request. |
| */ |
| protected void processNearSingleGetRequest(final UUID nodeId, final GridNearSingleGetRequest req) { |
| assert ctx.affinityNode(); |
| |
| final CacheExpiryPolicy expiryPlc = CacheExpiryPolicy.fromRemote(req.createTtl(), req.accessTtl()); |
| |
| IgniteInternalFuture<GridCacheEntryInfo> fut = |
| getDhtSingleAsync( |
| nodeId, |
| req.messageId(), |
| req.key(), |
| req.addReader(), |
| req.readThrough(), |
| req.topologyVersion(), |
| req.taskNameHash(), |
| expiryPlc, |
| req.skipValues(), |
| req.recovery(), |
| req.txLabel()); |
| |
| fut.listen(new CI1<IgniteInternalFuture<GridCacheEntryInfo>>() { |
| @Override public void apply(IgniteInternalFuture<GridCacheEntryInfo> f) { |
| GridNearSingleGetResponse res; |
| |
| GridDhtFuture<GridCacheEntryInfo> fut = (GridDhtFuture<GridCacheEntryInfo>)f; |
| |
| try { |
| GridCacheEntryInfo info = fut.get(); |
| |
| if (F.isEmpty(fut.invalidPartitions())) { |
| Message res0 = null; |
| |
| if (info != null) { |
| if (req.needEntryInfo()) { |
| info.key(null); |
| |
| res0 = info; |
| } |
| else if (req.needVersion()) |
| res0 = new CacheVersionedValue(info.value(), info.version()); |
| else |
| res0 = info.value(); |
| } |
| |
| res = new GridNearSingleGetResponse( |
| ctx.cacheId(), |
| req.futureId(), |
| null, |
| res0, |
| false, |
| req.addDeploymentInfo() |
| ); |
| |
| if (info != null && req.skipValues()) |
| res.setContainsValue(); |
| } |
| else { |
| AffinityTopologyVersion topVer = ctx.shared().exchange().lastTopologyFuture().initialVersion(); |
| |
| res = new GridNearSingleGetResponse( |
| ctx.cacheId(), |
| req.futureId(), |
| topVer, |
| null, |
| true, |
| req.addDeploymentInfo() |
| ); |
| } |
| } |
| catch (NodeStoppingException ignored) { |
| return; |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed processing get request: " + req, e); |
| |
| res = new GridNearSingleGetResponse(ctx.cacheId(), |
| req.futureId(), |
| req.topologyVersion(), |
| null, |
| false, |
| req.addDeploymentInfo()); |
| |
| res.error(e); |
| } |
| |
| try { |
| ctx.io().send(nodeId, res, ctx.ioPolicy()); |
| } |
| catch (ClusterTopologyCheckedException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send get response to node, node failed: " + nodeId); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send get response to node (is node still alive?) [nodeId=" + nodeId + |
| ",req=" + req + ", res=" + res + ']', e); |
| } |
| |
| sendTtlUpdateRequest(expiryPlc); |
| } |
| }); |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param req Get request. |
| */ |
| protected void processNearGetRequest(final UUID nodeId, final GridNearGetRequest req) { |
| assert ctx.affinityNode(); |
| |
| final CacheExpiryPolicy expiryPlc = CacheExpiryPolicy.fromRemote(req.createTtl(), req.accessTtl()); |
| |
| IgniteInternalFuture<Collection<GridCacheEntryInfo>> fut = |
| getDhtAsync(nodeId, |
| req.messageId(), |
| req.keys(), |
| req.addReaders(), |
| req.readThrough(), |
| req.topologyVersion(), |
| req.taskNameHash(), |
| expiryPlc, |
| req.skipValues(), |
| req.recovery(), |
| req.txLabel()); |
| |
| fut.listen(new CI1<IgniteInternalFuture<Collection<GridCacheEntryInfo>>>() { |
| @Override public void apply(IgniteInternalFuture<Collection<GridCacheEntryInfo>> f) { |
| GridNearGetResponse res = new GridNearGetResponse(ctx.cacheId(), |
| req.futureId(), |
| req.miniId(), |
| req.version(), |
| req.deployInfo() != null); |
| |
| GridDhtFuture<Collection<GridCacheEntryInfo>> fut = |
| (GridDhtFuture<Collection<GridCacheEntryInfo>>)f; |
| |
| try { |
| Collection<GridCacheEntryInfo> entries = fut.get(); |
| |
| res.entries(entries); |
| } |
| catch (NodeStoppingException ignored) { |
| return; |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed processing get request: " + req, e); |
| |
| res.error(e); |
| } |
| |
| if (!F.isEmpty(fut.invalidPartitions())) { |
| AffinityTopologyVersion topVer = ctx.shared().exchange().lastTopologyFuture().initialVersion(); |
| |
| res.invalidPartitions(fut.invalidPartitions(), topVer); |
| } |
| |
| try { |
| ctx.io().send(nodeId, res, ctx.ioPolicy()); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send get response to node (is node still alive?) [nodeId=" + nodeId + |
| ",req=" + req + ", res=" + res + ']', e); |
| } |
| |
| sendTtlUpdateRequest(expiryPlc); |
| } |
| }); |
| } |
| |
| /** |
| * Initiates process of notifying all interested nodes that TTL was changed. |
| * Directly sends requests to primary nodes and {@link IgniteCacheExpiryPolicy#readers()}. |
| * |
| * @param expiryPlc Expiry policy. |
| */ |
| public void sendTtlUpdateRequest(@Nullable final IgniteCacheExpiryPolicy expiryPlc) { |
| if (expiryPlc != null) |
| sendTtlUpdateRequest(expiryPlc, ctx.shared().exchange().readyAffinityVersion(), null); |
| } |
| |
| /** |
| * @param expiryPlc Expiry policy. |
| * @param topVer Topology version. |
| * @param srcNodeId ID of the node that sent the initial ttl request. |
| */ |
| private void sendTtlUpdateRequest( |
| IgniteCacheExpiryPolicy expiryPlc, |
| AffinityTopologyVersion topVer, |
| @Nullable UUID srcNodeId |
| ) { |
| if (!F.isEmpty(expiryPlc.entries())) { |
| ctx.closures().runLocalSafe(new GridPlainRunnable() { |
| @SuppressWarnings({"ForLoopReplaceableByForEach"}) |
| @Override public void run() { |
| Map<KeyCacheObject, GridCacheVersion> entries = expiryPlc.entries(); |
| |
| assert entries != null && !entries.isEmpty(); |
| |
| Map<ClusterNode, GridCacheTtlUpdateRequest> reqMap = new HashMap<>(); |
| |
| for (Map.Entry<KeyCacheObject, GridCacheVersion> e : entries.entrySet()) { |
| ClusterNode primaryNode = ctx.affinity().primaryByKey(e.getKey(), topVer); |
| |
| if (primaryNode.isLocal()) { |
| Collection<ClusterNode> nodes = ctx.affinity().backupsByKey(e.getKey(), topVer); |
| |
| for (Iterator<ClusterNode> nodesIter = nodes.iterator(); nodesIter.hasNext(); ) { |
| ClusterNode node = nodesIter.next(); |
| |
| // There's no need to re-send ttl update request to the node |
| // that sent the initial ttl update request. |
| if (srcNodeId != null && srcNodeId.equals(node.id())) |
| continue; |
| |
| GridCacheTtlUpdateRequest req = reqMap.get(node); |
| |
| if (req == null) { |
| reqMap.put(node, req = new GridCacheTtlUpdateRequest(ctx.cacheId(), |
| topVer, |
| expiryPlc.forAccess())); |
| } |
| |
| req.addEntry(e.getKey(), e.getValue()); |
| } |
| } |
| else { |
| // There is no need to re-send ttl update requests if we are not on the primary node. |
| if (srcNodeId != null) |
| continue; |
| |
| GridCacheTtlUpdateRequest req = reqMap.get(primaryNode); |
| |
| if (req == null) { |
| reqMap.put(primaryNode, req = new GridCacheTtlUpdateRequest(ctx.cacheId(), |
| topVer, |
| expiryPlc.forAccess())); |
| } |
| |
| req.addEntry(e.getKey(), e.getValue()); |
| } |
| } |
| |
| Map<UUID, Collection<IgniteBiTuple<KeyCacheObject, GridCacheVersion>>> rdrs = expiryPlc.readers(); |
| |
| if (rdrs != null) { |
| assert !rdrs.isEmpty(); |
| |
| for (Map.Entry<UUID, Collection<IgniteBiTuple<KeyCacheObject, GridCacheVersion>>> e : rdrs.entrySet()) { |
| ClusterNode node = ctx.node(e.getKey()); |
| |
| if (node != null) { |
| GridCacheTtlUpdateRequest req = reqMap.get(node); |
| |
| if (req == null) { |
| reqMap.put(node, req = new GridCacheTtlUpdateRequest(ctx.cacheId(), |
| topVer, |
| expiryPlc.forAccess())); |
| } |
| |
| for (IgniteBiTuple<KeyCacheObject, GridCacheVersion> t : e.getValue()) |
| req.addNearEntry(t.get1(), t.get2()); |
| } |
| } |
| } |
| |
| for (Map.Entry<ClusterNode, GridCacheTtlUpdateRequest> req : reqMap.entrySet()) { |
| try { |
| ctx.io().send(req.getKey(), req.getValue(), ctx.ioPolicy()); |
| } |
| catch (IgniteCheckedException e) { |
| if (e instanceof ClusterTopologyCheckedException) { |
| if (log.isDebugEnabled()) |
| log.debug("Failed to send TTC update request, node left: " + req.getKey()); |
| } |
| else |
| U.error(log, "Failed to send TTL update request.", e); |
| } |
| } |
| } |
| }); |
| } |
| } |
| |
| /** |
| * @param req Request. |
| */ |
| private void processTtlUpdateRequest(UUID srcNodeId, GridCacheTtlUpdateRequest req) { |
| final CacheExpiryPolicy expiryPlc = CacheExpiryPolicy.fromRemote(CU.TTL_NOT_CHANGED, req.ttl()); |
| |
| if (req.keys() != null) |
| updateTtl(this, req.keys(), req.versions(), expiryPlc); |
| |
| if (req.nearKeys() != null) { |
| GridNearCacheAdapter<K, V> near = near(); |
| |
| assert near != null; |
| |
| updateTtl(near, req.nearKeys(), req.nearVersions(), expiryPlc); |
| } |
| |
| sendTtlUpdateRequest(expiryPlc, req.topologyVersion(), srcNodeId); |
| } |
| |
| /** |
| * @param cache Cache. |
| * @param keys Entries keys. |
| * @param vers Entries versions. |
| * @param expiryPlc Expiry policy. |
| */ |
| private void updateTtl( |
| GridCacheAdapter<K, V> cache, |
| List<KeyCacheObject> keys, |
| List<GridCacheVersion> vers, |
| IgniteCacheExpiryPolicy expiryPlc |
| ) { |
| assert !F.isEmpty(keys); |
| assert keys.size() == vers.size(); |
| |
| int size = keys.size(); |
| |
| for (int i = 0; i < size; i++) { |
| try { |
| GridCacheEntryEx entry = null; |
| |
| try { |
| while (true) { |
| try { |
| entry = cache.entryEx(keys.get(i)); |
| |
| entry.unswap(false); |
| |
| entry.updateTtl(vers.get(i), expiryPlc); |
| |
| break; |
| } |
| catch (GridCacheEntryRemovedException ignore) { |
| if (log.isDebugEnabled()) |
| log.debug("Got removed entry: " + entry); |
| } |
| catch (GridDhtInvalidPartitionException e) { |
| if (log.isDebugEnabled()) |
| log.debug("Got GridDhtInvalidPartitionException: " + e); |
| |
| break; |
| } |
| } |
| } |
| finally { |
| if (entry != null) |
| entry.touch(); |
| } |
| } |
| catch (IgniteCheckedException e) { |
| log.error("Failed to unswap entry.", e); |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void unlockAll(Collection<? extends K> keys) { |
| assert false; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return S.toString(GridDhtCacheAdapter.class, this, super.toString()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public List<GridCacheClearAllRunnable<K, V>> splitClearLocally(boolean srv, boolean near, |
| boolean readers) { |
| return ctx.affinityNode() ? super.splitClearLocally(srv, near, readers) : |
| Collections.<GridCacheClearAllRunnable<K, V>>emptyList(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onDeferredDelete(GridCacheEntryEx entry, GridCacheVersion ver) { |
| assert entry.isDht(); |
| |
| GridDhtLocalPartition part = topology().localPartition(entry.partition(), AffinityTopologyVersion.NONE, |
| false); |
| |
| if (part != null) |
| part.onDeferredDelete(entry.context().cacheId(), entry.key(), ver); |
| } |
| |
| /** |
| * @param rmtVer Topology version that the cache request was mapped to by the remote node. |
| * @return {@code True} if cache affinity changed and operation should be remapped. |
| */ |
| protected final boolean needRemap(AffinityTopologyVersion rmtVer) { |
| // TODO IGNITE-7164 check mvcc crd for mvcc enabled txs. |
| return !ctx.affinity().isCompatibleWithCurrentTopologyVersion(rmtVer); |
| } |
| |
| /** |
| * @param primary If {@code true} includes primary entries. |
| * @param backup If {@code true} includes backup entries. |
| * @param keepBinary Keep binary flag. |
| * @return Local entries iterator. |
| */ |
| public Iterator<Cache.Entry<K, V>> localEntriesIterator(final boolean primary, |
| final boolean backup, |
| final boolean keepBinary) { |
| return localEntriesIterator(primary, |
| backup, |
| keepBinary, |
| ctx.affinity().affinityTopologyVersion()); |
| } |
| |
| /** |
| * @param primary If {@code true} includes primary entries. |
| * @param backup If {@code true} includes backup entries. |
| * @param keepBinary Keep binary flag. |
| * @param topVer Specified affinity topology version. |
| * @return Local entries iterator. |
| */ |
| private Iterator<Cache.Entry<K, V>> localEntriesIterator(final boolean primary, |
| final boolean backup, |
| final boolean keepBinary, |
| final AffinityTopologyVersion topVer) { |
| |
| return iterator(localEntriesIteratorEx(primary, backup, topVer), !keepBinary); |
| } |
| |
| /** |
| * @param primary If {@code true} includes primary entries. |
| * @param backup If {@code true} includes backup entries. |
| * @param topVer Specified affinity topology version. |
| * @return Local entries iterator. |
| */ |
| private Iterator<? extends GridCacheEntryEx> localEntriesIteratorEx(final boolean primary, |
| final boolean backup, |
| final AffinityTopologyVersion topVer) { |
| assert primary || backup; |
| |
| if (primary && backup) |
| return entries().iterator(); |
| else { |
| final Iterator<GridDhtLocalPartition> partIt = topology().currentLocalPartitions().iterator(); |
| |
| return new Iterator<GridCacheMapEntry>() { |
| private GridCacheMapEntry next; |
| |
| private Iterator<GridCacheMapEntry> curIt; |
| |
| { |
| advance(); |
| } |
| |
| @Override public boolean hasNext() { |
| return next != null; |
| } |
| |
| @Override public GridCacheMapEntry next() { |
| if (next == null) |
| throw new NoSuchElementException(); |
| |
| GridCacheMapEntry e = next; |
| |
| advance(); |
| |
| return e; |
| } |
| |
| @Override public void remove() { |
| throw new UnsupportedOperationException(); |
| } |
| |
| private void advance() { |
| next = null; |
| |
| do { |
| if (curIt == null) { |
| while (partIt.hasNext()) { |
| GridDhtLocalPartition part = partIt.next(); |
| |
| if (primary == part.primary(topVer)) { |
| curIt = part.entries(ctx.cacheId()).iterator(); |
| |
| break; |
| } |
| } |
| } |
| |
| if (curIt != null) { |
| if (curIt.hasNext()) { |
| next = curIt.next(); |
| |
| break; |
| } |
| else |
| curIt = null; |
| } |
| } |
| while (partIt.hasNext()); |
| } |
| }; |
| } |
| } |
| |
| /** |
| * Multi update future. |
| */ |
| private static class MultiUpdateFuture extends GridFutureAdapter<IgniteUuid> { |
| /** Topology version. */ |
| private AffinityTopologyVersion topVer; |
| |
| /** |
| * @param topVer Topology version. |
| */ |
| private MultiUpdateFuture(@NotNull AffinityTopologyVersion topVer) { |
| this.topVer = topVer; |
| } |
| |
| /** |
| * @return Topology version. |
| */ |
| private AffinityTopologyVersion topologyVersion() { |
| return topVer; |
| } |
| } |
| |
| /** |
| * |
| */ |
| protected abstract class MessageHandler<M> implements IgniteBiInClosure<UUID, M> { |
| /** */ |
| private static final long serialVersionUID = 0L; |
| |
| /** {@inheritDoc} */ |
| @Override public void apply(UUID nodeId, M msg) { |
| ClusterNode node = ctx.node(nodeId); |
| |
| if (node == null) { |
| if (log.isDebugEnabled()) |
| log.debug("Received message from failed node [node=" + nodeId + ", msg=" + msg + ']'); |
| |
| return; |
| } |
| |
| if (log.isDebugEnabled()) |
| log.debug("Received message from node [node=" + nodeId + ", msg=" + msg + ']'); |
| |
| onMessage(node, msg); |
| } |
| |
| /** |
| * @param node Node. |
| * @param msg Message. |
| */ |
| protected abstract void onMessage(ClusterNode node, M msg); |
| } |
| } |
