| /* |
| * 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.local.atomic; |
| |
| import java.io.Externalizable; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.Callable; |
| import javax.cache.expiry.ExpiryPolicy; |
| import javax.cache.processor.EntryProcessor; |
| import javax.cache.processor.EntryProcessorException; |
| import javax.cache.processor.EntryProcessorResult; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.cache.ReadRepairStrategy; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; |
| import org.apache.ignite.internal.processors.cache.CacheEntryPredicate; |
| import org.apache.ignite.internal.processors.cache.CacheInvokeEntry; |
| import org.apache.ignite.internal.processors.cache.CacheInvokeResult; |
| import org.apache.ignite.internal.processors.cache.CacheLazyEntry; |
| import org.apache.ignite.internal.processors.cache.CacheObject; |
| import org.apache.ignite.internal.processors.cache.CacheOperationContext; |
| import org.apache.ignite.internal.processors.cache.CachePartialUpdateCheckedException; |
| import org.apache.ignite.internal.processors.cache.CacheStorePartialUpdateException; |
| import org.apache.ignite.internal.processors.cache.EntryGetResult; |
| import org.apache.ignite.internal.processors.cache.GridCacheContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryEx; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException; |
| import org.apache.ignite.internal.processors.cache.GridCacheOperation; |
| import org.apache.ignite.internal.processors.cache.GridCachePreloader; |
| import org.apache.ignite.internal.processors.cache.GridCachePreloaderAdapter; |
| import org.apache.ignite.internal.processors.cache.GridCacheReturn; |
| import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy; |
| import org.apache.ignite.internal.processors.cache.KeyCacheObject; |
| import org.apache.ignite.internal.processors.cache.LockedEntriesInfo; |
| import org.apache.ignite.internal.processors.cache.local.GridLocalCache; |
| import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow; |
| import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx; |
| import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; |
| import org.apache.ignite.internal.processors.resource.GridResourceIoc; |
| import org.apache.ignite.internal.util.F0; |
| import org.apache.ignite.internal.util.future.GridFinishedFuture; |
| import org.apache.ignite.internal.util.lang.GridPlainCallable; |
| import org.apache.ignite.internal.util.lang.GridTuple3; |
| import org.apache.ignite.internal.util.typedef.C1; |
| import org.apache.ignite.internal.util.typedef.CI1; |
| import org.apache.ignite.internal.util.typedef.CX1; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.T2; |
| import org.apache.ignite.internal.util.typedef.internal.A; |
| import org.apache.ignite.internal.util.typedef.internal.CU; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteBiTuple; |
| import org.apache.ignite.plugin.security.SecurityPermission; |
| import org.apache.ignite.thread.IgniteThread; |
| import org.apache.ignite.transactions.TransactionIsolation; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.ignite.internal.processors.cache.GridCacheOperation.DELETE; |
| import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM; |
| import static org.apache.ignite.internal.processors.cache.GridCacheOperation.UPDATE; |
| |
| /** |
| * Non-transactional local cache. |
| */ |
| public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> { |
| /** */ |
| private static final long serialVersionUID = 0L; |
| |
| /** */ |
| private GridCachePreloader preldr; |
| |
| /** Locked entries info for each thread. */ |
| private final LockedEntriesInfo lockedEntriesInfo = new LockedEntriesInfo(); |
| |
| /** |
| * Empty constructor required by {@link Externalizable}. |
| */ |
| public GridLocalAtomicCache() { |
| // No-op. |
| } |
| |
| /** |
| * @param ctx Cache context. |
| */ |
| public GridLocalAtomicCache(GridCacheContext<K, V> ctx) { |
| super(ctx); |
| |
| preldr = new GridCachePreloaderAdapter(ctx.group()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void checkJta() throws IgniteCheckedException { |
| // No-op. |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean isLocal() { |
| return true; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public GridCachePreloader preloader() { |
| return preldr; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected V getAndPut0(K key, V val, @Nullable CacheEntryPredicate filter) throws IgniteCheckedException { |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| return (V)updateAllInternal(UPDATE, |
| Collections.singleton(key), |
| Collections.singleton(val), |
| null, |
| expiryPerCall(), |
| true, |
| false, |
| filter, |
| ctx.writeThrough(), |
| ctx.readThrough(), |
| opCtx != null && opCtx.isKeepBinary()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected boolean put0(K key, V val, CacheEntryPredicate filter) throws IgniteCheckedException { |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| Boolean res = (Boolean)updateAllInternal(UPDATE, |
| Collections.singleton(key), |
| Collections.singleton(val), |
| null, |
| expiryPerCall(), |
| false, |
| false, |
| filter, |
| ctx.writeThrough(), |
| ctx.readThrough(), |
| opCtx != null && opCtx.isKeepBinary()); |
| |
| assert res != null; |
| |
| return res; |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("unchecked") |
| @Override public IgniteInternalFuture<V> getAndPutAsync0(K key, V val, @Nullable CacheEntryPredicate filter) { |
| return updateAllAsync0(F0.asMap(key, val), |
| null, |
| null, |
| true, |
| false, |
| filter); |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("unchecked") |
| @Override public IgniteInternalFuture<Boolean> putAsync0(K key, V val, @Nullable CacheEntryPredicate filter) { |
| return updateAllAsync0(F0.asMap(key, val), |
| null, |
| null, |
| false, |
| false, |
| filter); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void putAll0(Map<? extends K, ? extends V> m) throws IgniteCheckedException { |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| updateAllInternal(UPDATE, |
| m.keySet(), |
| m.values(), |
| null, |
| expiryPerCall(), |
| false, |
| false, |
| null, |
| ctx.writeThrough(), |
| ctx.readThrough(), |
| opCtx != null && opCtx.isKeepBinary()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<?> putAllAsync0(Map<? extends K, ? extends V> m) { |
| return updateAllAsync0(m, |
| null, |
| null, |
| false, |
| false, |
| null).chain(RET2NULL); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected V getAndRemove0(K key) throws IgniteCheckedException { |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| return (V)updateAllInternal(DELETE, |
| Collections.singleton(key), |
| null, |
| null, |
| expiryPerCall(), |
| true, |
| false, |
| null, |
| ctx.writeThrough(), |
| ctx.readThrough(), |
| opCtx != null && opCtx.isKeepBinary()); |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("unchecked") |
| @Override public IgniteInternalFuture<V> getAndRemoveAsync0(K key) { |
| return removeAllAsync0(Collections.singletonList(key), true, false, null); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void removeAll0(Collection<? extends K> keys) throws IgniteCheckedException { |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| updateAllInternal(DELETE, |
| keys, |
| null, |
| null, |
| expiryPerCall(), |
| false, |
| false, |
| null, |
| ctx.writeThrough(), |
| ctx.readThrough(), |
| opCtx != null && opCtx.isKeepBinary()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<Object> removeAllAsync0(Collection<? extends K> keys) { |
| return removeAllAsync0(keys, false, false, null).chain(RET2NULL); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean remove0(K key, final CacheEntryPredicate filter) throws IgniteCheckedException { |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| Boolean rmv = (Boolean)updateAllInternal(DELETE, |
| Collections.singleton(key), |
| null, |
| null, |
| expiryPerCall(), |
| false, |
| false, |
| filter, |
| ctx.writeThrough(), |
| ctx.readThrough(), |
| opCtx != null && opCtx.isKeepBinary()); |
| |
| assert rmv != null; |
| |
| return rmv; |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("unchecked") |
| @Override public IgniteInternalFuture<Boolean> removeAsync0(K key, @Nullable CacheEntryPredicate filter) { |
| return removeAllAsync0(Collections.singletonList(key), false, false, filter); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<?> removeAllAsync() { |
| return ctx.closures().callLocalSafe(new GridPlainCallable<Void>() { |
| @Override public Void call() throws Exception { |
| removeAll(); |
| |
| return null; |
| } |
| }); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected V get( |
| final K key, |
| String taskName, |
| boolean deserializeBinary, |
| boolean needVer) throws IgniteCheckedException { |
| Map<K, V> m = getAllInternal(Collections.singleton(key), |
| ctx.readThrough(), |
| taskName, |
| deserializeBinary, |
| false, |
| needVer); |
| |
| assert m.isEmpty() || m.size() == 1 : m.size(); |
| |
| return F.firstValue(m); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public final Map<K, V> getAll( |
| Collection<? extends K> keys, |
| boolean deserializeBinary, |
| boolean needVer, |
| boolean recovery, |
| ReadRepairStrategy readRepairStrategy) throws IgniteCheckedException { |
| A.notNull(keys, "keys"); |
| |
| String taskName = ctx.kernalContext().job().currentTaskName(); |
| |
| return getAllInternal(keys, |
| ctx.readThrough(), |
| taskName, |
| deserializeBinary, |
| false, |
| needVer); |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("unchecked") |
| @Override public IgniteInternalFuture<Map<K, V>> getAllAsync( |
| @Nullable final Collection<? extends K> keys, |
| final boolean forcePrimary, |
| boolean skipTx, |
| final String taskName, |
| final boolean deserializeBinary, |
| boolean recovery, |
| ReadRepairStrategy readRepairStrategy, |
| final boolean skipVals, |
| final boolean needVer |
| ) { |
| A.notNull(keys, "keys"); |
| |
| final boolean storeEnabled = ctx.readThrough(); |
| |
| return asyncOp(new GridPlainCallable<Map<K, V>>() { |
| @Override public Map<K, V> call() throws Exception { |
| return getAllInternal(keys, storeEnabled, taskName, deserializeBinary, skipVals, needVer); |
| } |
| }); |
| } |
| |
| /** |
| * Entry point to all public API get methods. |
| * |
| * @param keys Keys to remove. |
| * @param storeEnabled Store enabled flag. |
| * @param taskName Task name. |
| * @param deserializeBinary Deserialize binary . |
| * @param skipVals Skip value flag. |
| * @param needVer Need version. |
| * @return Key-value map. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @SuppressWarnings("ConstantConditions") |
| private Map<K, V> getAllInternal(@Nullable Collection<? extends K> keys, |
| boolean storeEnabled, |
| String taskName, |
| boolean deserializeBinary, |
| boolean skipVals, |
| boolean needVer |
| ) throws IgniteCheckedException { |
| ctx.checkSecurity(SecurityPermission.CACHE_READ); |
| |
| if (F.isEmpty(keys)) |
| return Collections.emptyMap(); |
| |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| Map<K, V> vals = U.newHashMap(keys.size()); |
| |
| warnIfUnordered(keys, BulkOperation.GET); |
| |
| final IgniteCacheExpiryPolicy expiry = expiryPolicy(opCtx != null ? opCtx.expiry() : null); |
| |
| boolean success = true; |
| boolean readNoEntry = ctx.readNoEntry(expiry, false); |
| final boolean evt = !skipVals; |
| |
| ctx.shared().database().checkpointReadLock(); |
| |
| try { |
| for (K key : keys) { |
| if (key == null) |
| throw new NullPointerException("Null key."); |
| |
| KeyCacheObject cacheKey = ctx.toCacheKeyObject(key); |
| |
| boolean skipEntry = readNoEntry; |
| |
| if (readNoEntry) { |
| CacheDataRow row = ctx.offheap().read(ctx, cacheKey); |
| |
| if (row != null) { |
| long expireTime = row.expireTime(); |
| |
| if (expireTime == 0 || expireTime > U.currentTimeMillis()) { |
| ctx.addResult(vals, |
| cacheKey, |
| row.value(), |
| skipVals, |
| false, |
| deserializeBinary, |
| true, |
| null, |
| row.version(), |
| 0, |
| 0, |
| needVer, |
| null); |
| |
| if (ctx.statisticsEnabled() && !skipVals) |
| metrics0().onRead(true); |
| |
| if (evt) { |
| ctx.events().readEvent(cacheKey, |
| null, |
| null, |
| row.value(), |
| taskName, |
| !deserializeBinary); |
| } |
| } |
| else |
| skipEntry = false; |
| } |
| else |
| success = false; |
| } |
| |
| if (!skipEntry) { |
| GridCacheEntryEx entry = null; |
| |
| while (true) { |
| try { |
| entry = entryEx(cacheKey); |
| |
| if (entry != null) { |
| CacheObject v; |
| |
| if (needVer) { |
| EntryGetResult res = entry.innerGetVersioned( |
| null, |
| null, |
| /*update-metrics*/false, |
| /*event*/evt, |
| null, |
| taskName, |
| expiry, |
| !deserializeBinary, |
| null); |
| |
| if (res != null) { |
| ctx.addResult( |
| vals, |
| cacheKey, |
| res, |
| skipVals, |
| false, |
| deserializeBinary, |
| true, |
| needVer); |
| } |
| else |
| success = false; |
| } |
| else { |
| v = entry.innerGet( |
| null, |
| null, |
| /*read-through*/false, |
| /*update-metrics*/true, |
| /*event*/evt, |
| null, |
| taskName, |
| expiry, |
| !deserializeBinary); |
| |
| if (v != null) { |
| ctx.addResult(vals, |
| cacheKey, |
| v, |
| skipVals, |
| false, |
| deserializeBinary, |
| true, |
| null, |
| 0, |
| 0, |
| null); |
| } |
| else |
| success = false; |
| } |
| } |
| |
| break; // While. |
| } |
| catch (GridCacheEntryRemovedException ignored) { |
| // No-op, retry. |
| } |
| finally { |
| if (entry != null) |
| entry.touch(); |
| } |
| |
| if (!success && storeEnabled) |
| break; |
| } |
| } |
| if (!success) { |
| if (!storeEnabled && ctx.statisticsEnabled() && !skipVals) |
| metrics0().onRead(false); |
| } |
| } |
| } |
| finally { |
| ctx.shared().database().checkpointReadUnlock(); |
| } |
| |
| if (success || !storeEnabled) |
| return vals; |
| |
| return getAllAsync( |
| keys, |
| null, |
| opCtx == null || !opCtx.skipStore(), |
| false, |
| taskName, |
| deserializeBinary, |
| opCtx != null && opCtx.recovery(), |
| null, |
| /*force primary*/false, |
| expiry, |
| skipVals, |
| needVer).get(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public <T> EntryProcessorResult<T> invoke(K key, |
| EntryProcessor<K, V, T> entryProcessor, |
| Object... args) throws IgniteCheckedException { |
| return invokeAsync(key, entryProcessor, args).get(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll(Set<? extends K> keys, |
| final EntryProcessor<K, V, T> entryProcessor, |
| Object... args) throws IgniteCheckedException { |
| A.notNull(keys, "keys", entryProcessor, "entryProcessor"); |
| |
| warnIfUnordered(keys, BulkOperation.INVOKE); |
| |
| final boolean statsEnabled = ctx.statisticsEnabled(); |
| |
| final long start = statsEnabled ? System.nanoTime() : 0L; |
| |
| Map<? extends K, EntryProcessor> invokeMap = F.viewAsMap(keys, new C1<K, EntryProcessor>() { |
| @Override public EntryProcessor apply(K k) { |
| return entryProcessor; |
| } |
| }); |
| |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| final boolean keepBinary = opCtx != null && opCtx.isKeepBinary(); |
| |
| Map<K, EntryProcessorResult<T>> entryProcessorRes = (Map<K, EntryProcessorResult<T>>)updateAllInternal( |
| TRANSFORM, |
| invokeMap.keySet(), |
| invokeMap.values(), |
| args, |
| expiryPerCall(), |
| false, |
| false, |
| null, |
| ctx.writeThrough(), |
| ctx.readThrough(), |
| keepBinary); |
| |
| if (statsEnabled) |
| metrics0().addInvokeTimeNanos(System.nanoTime() - start); |
| |
| return entryProcessorRes; |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("unchecked") |
| @Override public <T> IgniteInternalFuture<EntryProcessorResult<T>> invokeAsync(K key, |
| EntryProcessor<K, V, T> entryProcessor, |
| Object... args) throws EntryProcessorException { |
| A.notNull(key, "key", entryProcessor, "entryProcessor"); |
| |
| final boolean statsEnabled = ctx.statisticsEnabled(); |
| |
| final long start = statsEnabled ? System.nanoTime() : 0L; |
| |
| Map<? extends K, EntryProcessor> invokeMap = |
| Collections.singletonMap(key, (EntryProcessor)entryProcessor); |
| |
| IgniteInternalFuture<Map<K, EntryProcessorResult<T>>> fut = updateAllAsync0(null, |
| invokeMap, |
| args, |
| false, |
| false, |
| null); |
| |
| return fut.chain(new CX1<IgniteInternalFuture<Map<K, EntryProcessorResult<T>>>, EntryProcessorResult<T>>() { |
| @Override public EntryProcessorResult<T> applyx(IgniteInternalFuture<Map<K, EntryProcessorResult<T>>> fut) |
| throws IgniteCheckedException { |
| Map<K, EntryProcessorResult<T>> resMap = fut.get(); |
| |
| if (statsEnabled) |
| metrics0().addInvokeTimeNanos(System.nanoTime() - start); |
| |
| if (resMap != null) { |
| assert resMap.isEmpty() || resMap.size() == 1 : resMap.size(); |
| |
| return resMap.isEmpty() ? new CacheInvokeResult<>() : resMap.values().iterator().next(); |
| } |
| |
| return new CacheInvokeResult<>(); |
| } |
| }); |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("unchecked") |
| @Override public <T> IgniteInternalFuture<Map<K, EntryProcessorResult<T>>> invokeAllAsync( |
| Set<? extends K> keys, |
| final EntryProcessor<K, V, T> entryProcessor, |
| Object... args) { |
| A.notNull(keys, "keys", entryProcessor, "entryProcessor"); |
| |
| warnIfUnordered(keys, BulkOperation.INVOKE); |
| |
| final boolean statsEnabled = ctx.statisticsEnabled(); |
| |
| final long start = statsEnabled ? System.nanoTime() : 0L; |
| |
| Map<? extends K, EntryProcessor> invokeMap = F.viewAsMap(keys, new C1<K, EntryProcessor>() { |
| @Override public EntryProcessor apply(K k) { |
| return entryProcessor; |
| } |
| }); |
| |
| IgniteInternalFuture fut = updateAllAsync0(null, |
| invokeMap, |
| args, |
| true, |
| false, |
| null); |
| |
| if (statsEnabled) |
| fut.listen(new InvokeAllTimeStatClosure(metrics0(), start)); |
| |
| return fut; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public <T> Map<K, EntryProcessorResult<T>> invokeAll( |
| Map<? extends K, ? extends EntryProcessor<K, V, T>> map, |
| Object... args) throws IgniteCheckedException { |
| A.notNull(map, "map"); |
| |
| warnIfUnordered(map, BulkOperation.INVOKE); |
| |
| final boolean statsEnabled = ctx.statisticsEnabled(); |
| |
| final long start = statsEnabled ? System.nanoTime() : 0L; |
| |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| Map<K, EntryProcessorResult<T>> entryProcessorResult = (Map<K, EntryProcessorResult<T>>)updateAllInternal( |
| TRANSFORM, |
| map.keySet(), |
| map.values(), |
| args, |
| expiryPerCall(), |
| false, |
| false, |
| null, |
| ctx.writeThrough(), |
| ctx.readThrough(), |
| opCtx != null && opCtx.isKeepBinary()); |
| |
| if (statsEnabled) |
| metrics0().addInvokeTimeNanos(System.nanoTime() - start); |
| |
| return entryProcessorResult; |
| } |
| |
| /** {@inheritDoc} */ |
| @SuppressWarnings("unchecked") |
| @Override public <T> IgniteInternalFuture<Map<K, EntryProcessorResult<T>>> invokeAllAsync( |
| Map<? extends K, ? extends EntryProcessor<K, V, T>> map, |
| Object... args) { |
| A.notNull(map, "map"); |
| |
| warnIfUnordered(map, BulkOperation.INVOKE); |
| |
| final boolean statsEnabled = ctx.statisticsEnabled(); |
| |
| final long start = statsEnabled ? System.nanoTime() : 0L; |
| |
| IgniteInternalFuture fut = updateAllAsync0(null, |
| map, |
| args, |
| true, |
| false, |
| null); |
| |
| if (statsEnabled) |
| fut.listen(new InvokeAllTimeStatClosure(metrics0(), start)); |
| |
| return fut; |
| } |
| |
| /** |
| * Entry point for public API update methods. |
| * |
| * @param map Put map. Either {@code map} or {@code invokeMap} should be passed. |
| * @param invokeMap Transform map. Either {@code map} or {@code invokeMap} should be passed. |
| * @param invokeArgs Optional arguments for EntryProcessor. |
| * @param retval Return value required flag. |
| * @param rawRetval Return {@code GridCacheReturn} instance. |
| * @param filter Cache entry filter for atomic updates. |
| * @return Completion future. |
| */ |
| private IgniteInternalFuture updateAllAsync0( |
| @Nullable final Map<? extends K, ? extends V> map, |
| @Nullable final Map<? extends K, ? extends EntryProcessor> invokeMap, |
| @Nullable final Object[] invokeArgs, |
| final boolean retval, |
| final boolean rawRetval, |
| @Nullable final CacheEntryPredicate filter |
| ) { |
| final GridCacheOperation op = invokeMap != null ? TRANSFORM : UPDATE; |
| |
| final Collection<? extends K> keys = |
| map != null ? map.keySet() : invokeMap != null ? invokeMap.keySet() : null; |
| |
| final Collection<?> vals = map != null ? map.values() : invokeMap != null ? invokeMap.values() : null; |
| |
| final boolean writeThrough = ctx.writeThrough(); |
| |
| final boolean readThrough = ctx.readThrough(); |
| |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| final ExpiryPolicy expiry = expiryPerCall(); |
| |
| final boolean keepBinary = opCtx != null && opCtx.isKeepBinary(); |
| |
| return asyncOp(new GridPlainCallable<Object>() { |
| @Override public Object call() throws Exception { |
| return updateAllInternal(op, |
| keys, |
| vals, |
| invokeArgs, |
| expiry, |
| retval, |
| rawRetval, |
| filter, |
| writeThrough, |
| readThrough, |
| keepBinary); |
| } |
| }); |
| } |
| |
| /** |
| * Entry point for public API remove methods. |
| * |
| * @param keys Keys to remove. |
| * @param retval Return value required flag. |
| * @param rawRetval Return {@code GridCacheReturn} instance. |
| * @param filter Cache entry filter. |
| * @return Completion future. |
| */ |
| private IgniteInternalFuture removeAllAsync0( |
| @Nullable final Collection<? extends K> keys, |
| final boolean retval, |
| final boolean rawRetval, |
| @Nullable final CacheEntryPredicate filter |
| ) { |
| final boolean writeThrough = ctx.writeThrough(); |
| |
| final boolean readThrough = ctx.readThrough(); |
| |
| final ExpiryPolicy expiryPlc = expiryPerCall(); |
| |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| final boolean keepBinary = opCtx != null && opCtx.isKeepBinary(); |
| |
| return asyncOp(new GridPlainCallable<Object>() { |
| @Override public Object call() throws Exception { |
| return updateAllInternal(DELETE, |
| keys, |
| null, |
| null, |
| expiryPlc, |
| retval, |
| rawRetval, |
| filter, |
| writeThrough, |
| readThrough, |
| keepBinary); |
| } |
| }); |
| } |
| |
| /** |
| * Entry point for all public update methods (put, remove, invoke). |
| * |
| * @param op Operation. |
| * @param keys Keys. |
| * @param vals Values. |
| * @param invokeArgs Optional arguments for EntryProcessor. |
| * @param expiryPlc Expiry policy. |
| * @param retval Return value required flag. |
| * @param rawRetval Return {@code GridCacheReturn} instance. |
| * @param filter Cache entry filter. |
| * @param writeThrough Write through. |
| * @param readThrough Read through. |
| * @return Update result. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @SuppressWarnings("unchecked") |
| private Object updateAllInternal(GridCacheOperation op, |
| Collection<? extends K> keys, |
| @Nullable Iterable<?> vals, |
| @Nullable Object[] invokeArgs, |
| @Nullable ExpiryPolicy expiryPlc, |
| boolean retval, |
| boolean rawRetval, |
| CacheEntryPredicate filter, |
| boolean writeThrough, |
| boolean readThrough, |
| boolean keepBinary |
| ) throws IgniteCheckedException { |
| if (op == DELETE) |
| ctx.checkSecurity(SecurityPermission.CACHE_REMOVE); |
| else |
| ctx.checkSecurity(SecurityPermission.CACHE_PUT); |
| |
| String taskName = ctx.kernalContext().job().currentTaskName(); |
| |
| GridCacheVersion ver = nextVersion(); |
| |
| CacheEntryPredicate[] filters = CU.filterArray(filter); |
| |
| IgniteBiTuple<Boolean, ?> res = null; |
| |
| CachePartialUpdateCheckedException err = null; |
| |
| ctx.shared().database().checkpointReadLock(); |
| |
| try { |
| ctx.shared().database().ensureFreeSpace(ctx.dataRegion()); |
| |
| if (writeThrough && keys.size() > 1) { |
| return updateWithBatch(op, |
| keys, |
| vals, |
| invokeArgs, |
| expiryPlc, |
| ver, |
| filters, |
| keepBinary, |
| taskName); |
| } |
| |
| Iterator<?> valsIter = vals != null ? vals.iterator() : null; |
| |
| boolean intercept = ctx.config().getInterceptor() != null; |
| |
| for (K key : keys) { |
| if (key == null) |
| throw new NullPointerException("Null key."); |
| |
| Object val = valsIter != null ? valsIter.next() : null; |
| |
| if (val == null && op != DELETE) |
| throw new NullPointerException("Null value."); |
| |
| KeyCacheObject cacheKey = ctx.toCacheKeyObject(key); |
| |
| if (op == UPDATE) { |
| val = ctx.toCacheObject(val); |
| |
| ctx.validateKeyAndValue(cacheKey, (CacheObject)val); |
| } |
| else if (op == TRANSFORM) |
| ctx.kernalContext().resource().inject(val, GridResourceIoc.AnnotationSet.ENTRY_PROCESSOR, ctx.name()); |
| |
| while (true) { |
| GridCacheEntryEx entry = null; |
| |
| try { |
| entry = entryEx(cacheKey); |
| |
| GridTuple3<Boolean, Object, EntryProcessorResult<Object>> t = entry.innerUpdateLocal( |
| ver, |
| val == null ? DELETE : op, |
| val, |
| invokeArgs, |
| writeThrough, |
| readThrough, |
| retval, |
| keepBinary, |
| expiryPlc, |
| true, |
| true, |
| filters, |
| intercept, |
| taskName, |
| false); |
| |
| if (op == TRANSFORM) { |
| if (t.get3() != null) { |
| Map<K, EntryProcessorResult> computedMap; |
| |
| if (res == null) { |
| computedMap = U.newHashMap(keys.size()); |
| |
| res = new IgniteBiTuple<>(true, computedMap); |
| } |
| else |
| computedMap = (Map<K, EntryProcessorResult>)res.get2(); |
| |
| computedMap.put(key, t.get3()); |
| } |
| } |
| else if (res == null) |
| res = new T2(t.get1(), t.get2()); |
| |
| break; // While. |
| } |
| catch (GridCacheEntryRemovedException ignored) { |
| if (log.isDebugEnabled()) |
| log.debug("Got removed entry while updating (will retry): " + key); |
| |
| entry = null; |
| } |
| catch (IgniteCheckedException e) { |
| if (err == null) |
| err = partialUpdateException(); |
| |
| err.add(F.asList(key), e); |
| |
| U.error(log, "Failed to update key : " + key, e); |
| |
| break; |
| } |
| finally { |
| if (entry != null) |
| entry.touch(); |
| } |
| } |
| } |
| } |
| finally { |
| ctx.shared().database().checkpointReadUnlock(); |
| } |
| |
| if (err != null) |
| throw err; |
| |
| Object ret = res == null ? null : rawRetval ? new GridCacheReturn( |
| ctx, |
| true, |
| keepBinary, |
| U.deploymentClassLoader(ctx.kernalContext(), U.contextDeploymentClassLoaderId(ctx.kernalContext())), |
| res.get2(), |
| res.get1() |
| ) : (retval || op == TRANSFORM) ? res.get2() : res.get1(); |
| |
| if (op == TRANSFORM && ret == null) |
| ret = Collections.emptyMap(); |
| |
| return ret; |
| } |
| |
| /** |
| * Updates entries using batched write-through. |
| * |
| * @param op Operation. |
| * @param keys Keys. |
| * @param vals Values. |
| * @param invokeArgs Optional arguments for EntryProcessor. |
| * @param expiryPlc Expiry policy. |
| * @param ver Cache version. |
| * @param filter Optional filter. |
| * @param taskName Task name. |
| * @return Results map for invoke operation. |
| * @throws CachePartialUpdateCheckedException If update failed. |
| */ |
| @SuppressWarnings({"ForLoopReplaceableByForEach", "unchecked"}) |
| private Map<K, EntryProcessorResult> updateWithBatch( |
| GridCacheOperation op, |
| Collection<? extends K> keys, |
| @Nullable Iterable<?> vals, |
| @Nullable Object[] invokeArgs, |
| @Nullable ExpiryPolicy expiryPlc, |
| GridCacheVersion ver, |
| @Nullable CacheEntryPredicate[] filter, |
| boolean keepBinary, |
| String taskName |
| ) throws IgniteCheckedException { |
| List<GridCacheEntryEx> locked = lockEntries(keys); |
| |
| try { |
| int size = locked.size(); |
| |
| Map<KeyCacheObject, CacheObject> putMap = null; |
| |
| Collection<KeyCacheObject> rmvKeys = null; |
| |
| List<CacheObject> writeVals = null; |
| |
| Map<K, EntryProcessorResult> invokeResMap = |
| op == TRANSFORM ? U.<K, EntryProcessorResult>newHashMap(size) : null; |
| |
| List<GridCacheEntryEx> filtered = new ArrayList<>(size); |
| |
| CachePartialUpdateCheckedException err = null; |
| |
| Iterator<?> valsIter = vals != null ? vals.iterator() : null; |
| |
| boolean intercept = ctx.config().getInterceptor() != null; |
| |
| for (int i = 0; i < size; i++) { |
| GridCacheEntryEx entry = locked.get(i); |
| |
| Object val = valsIter != null ? valsIter.next() : null; |
| |
| if (val == null && op != DELETE) |
| throw new NullPointerException("Null value."); |
| |
| try { |
| try { |
| if (!ctx.isAllLocked(entry, filter)) { |
| if (log.isDebugEnabled()) |
| log.debug("Entry did not pass the filter (will skip write) [entry=" + entry + |
| ", filter=" + Arrays.toString(filter) + ']'); |
| |
| continue; |
| } |
| } |
| catch (IgniteCheckedException e) { |
| if (err == null) |
| err = partialUpdateException(); |
| |
| err.add(F.asList(entry.key()), e); |
| |
| continue; |
| } |
| |
| if (op == TRANSFORM) { |
| ctx.kernalContext().resource().inject(val, |
| GridResourceIoc.AnnotationSet.ENTRY_PROCESSOR, |
| ctx.name()); |
| |
| EntryProcessor<Object, Object, Object> entryProcessor = |
| (EntryProcessor<Object, Object, Object>)val; |
| |
| CacheObject old = entry.innerGet( |
| null, |
| null, |
| /*read-through*/true, |
| /*update-metrics*/true, |
| /*event*/true, |
| entryProcessor, |
| taskName, |
| null, |
| keepBinary); |
| |
| Object oldVal = null; |
| |
| CacheInvokeEntry<Object, Object> invokeEntry = new CacheInvokeEntry<>(entry.key(), old, |
| entry.version(), keepBinary, entry); |
| |
| CacheObject updated; |
| Object updatedVal = null; |
| CacheInvokeResult invokeRes = null; |
| |
| boolean validation = false; |
| |
| IgniteThread.onEntryProcessorEntered(false); |
| |
| try { |
| Object computed = entryProcessor.process(invokeEntry, invokeArgs); |
| |
| updatedVal = ctx.unwrapTemporary(invokeEntry.getValue()); |
| |
| updated = ctx.toCacheObject(updatedVal); |
| |
| if (computed != null) |
| invokeRes = CacheInvokeResult.fromResult(ctx.unwrapTemporary(computed)); |
| |
| if (invokeEntry.modified() && updated != null) { |
| validation = true; |
| |
| ctx.validateKeyAndValue(entry.key(), updated); |
| } |
| else if (ctx.statisticsEnabled() && !invokeEntry.modified()) |
| ctx.cache().metrics0().onReadOnlyInvoke(old != null); |
| } |
| catch (Exception e) { |
| invokeRes = CacheInvokeResult.fromError(e); |
| |
| updated = old; |
| |
| if (validation) { |
| invokeResMap.put((K)entry.key().value(ctx.cacheObjectContext(), false), invokeRes); |
| |
| continue; |
| } |
| } |
| finally { |
| IgniteThread.onEntryProcessorLeft(); |
| } |
| |
| if (invokeRes != null) |
| invokeResMap.put((K)entry.key().value(ctx.cacheObjectContext(), false), invokeRes); |
| |
| if (updated == null) { |
| if (intercept) { |
| IgniteBiTuple<Boolean, ?> interceptorRes = ctx.config().getInterceptor() |
| .onBeforeRemove(new CacheLazyEntry(ctx, entry.key(), invokeEntry.key(), |
| old, oldVal, keepBinary)); |
| |
| if (ctx.cancelRemove(interceptorRes)) |
| continue; |
| } |
| |
| // Update previous batch. |
| if (putMap != null) { |
| err = updatePartialBatch( |
| filtered, |
| ver, |
| writeVals, |
| putMap, |
| null, |
| expiryPlc, |
| keepBinary, |
| err, |
| taskName, |
| true); |
| |
| putMap = null; |
| writeVals = null; |
| |
| filtered = new ArrayList<>(); |
| } |
| |
| // Start collecting new batch. |
| if (rmvKeys == null) |
| rmvKeys = new ArrayList<>(size); |
| |
| rmvKeys.add(entry.key()); |
| } |
| else { |
| if (intercept) { |
| Object interceptorVal = ctx.config().getInterceptor() |
| .onBeforePut(new CacheLazyEntry(ctx, entry.key(), invokeEntry.getKey(), |
| old, oldVal, keepBinary), updatedVal); |
| |
| if (interceptorVal == null) |
| continue; |
| |
| updated = ctx.toCacheObject(ctx.unwrapTemporary(interceptorVal)); |
| } |
| |
| // Update previous batch. |
| if (rmvKeys != null) { |
| err = updatePartialBatch( |
| filtered, |
| ver, |
| null, |
| null, |
| rmvKeys, |
| expiryPlc, |
| keepBinary, |
| err, |
| taskName, |
| true); |
| |
| rmvKeys = null; |
| |
| filtered = new ArrayList<>(); |
| } |
| |
| if (putMap == null) { |
| putMap = new LinkedHashMap<>(size, 1.0f); |
| writeVals = new ArrayList<>(size); |
| } |
| |
| putMap.put(entry.key(), updated); |
| writeVals.add(updated); |
| } |
| } |
| else if (op == UPDATE) { |
| CacheObject cacheVal = ctx.toCacheObject(val); |
| |
| if (intercept) { |
| CacheObject old = entry.innerGet( |
| null, |
| null, |
| /*read-through*/ctx.loadPreviousValue(), |
| /*update-metrics*/true, |
| /*event*/true, |
| null, |
| taskName, |
| null, |
| keepBinary); |
| |
| Object interceptorVal = ctx.config().getInterceptor().onBeforePut(new CacheLazyEntry( |
| ctx, entry.key(), old, keepBinary), val); |
| |
| if (interceptorVal == null) |
| continue; |
| |
| cacheVal = ctx.toCacheObject(ctx.unwrapTemporary(interceptorVal)); |
| } |
| |
| ctx.validateKeyAndValue(entry.key(), cacheVal); |
| |
| if (putMap == null) { |
| putMap = new LinkedHashMap<>(size, 1.0f); |
| writeVals = new ArrayList<>(size); |
| } |
| |
| putMap.put(entry.key(), cacheVal); |
| writeVals.add(cacheVal); |
| } |
| else { |
| assert op == DELETE; |
| |
| if (intercept) { |
| CacheObject old = entry.innerGet( |
| null, |
| null, |
| /*read-through*/ctx.loadPreviousValue(), |
| /*update-metrics*/true, |
| /*event*/true, |
| null, |
| taskName, |
| null, |
| keepBinary); |
| |
| IgniteBiTuple<Boolean, ?> interceptorRes = ctx.config().getInterceptor() |
| .onBeforeRemove(new CacheLazyEntry(ctx, entry.key(), old, keepBinary)); |
| |
| if (ctx.cancelRemove(interceptorRes)) |
| continue; |
| } |
| |
| if (rmvKeys == null) |
| rmvKeys = new ArrayList<>(size); |
| |
| rmvKeys.add(entry.key()); |
| } |
| |
| filtered.add(entry); |
| } |
| catch (IgniteCheckedException e) { |
| if (err == null) |
| err = partialUpdateException(); |
| |
| err.add(F.asList(entry.key()), e); |
| } |
| catch (GridCacheEntryRemovedException ignore) { |
| assert false : "Entry cannot become obsolete while holding lock."; |
| } |
| } |
| |
| // Store final batch. |
| if (putMap != null || rmvKeys != null) { |
| err = updatePartialBatch( |
| filtered, |
| ver, |
| writeVals, |
| putMap, |
| rmvKeys, |
| expiryPlc, |
| keepBinary, |
| err, |
| taskName, |
| op == TRANSFORM); |
| } |
| else |
| assert filtered.isEmpty(); |
| |
| if (err != null) |
| throw err; |
| |
| return invokeResMap; |
| } |
| finally { |
| unlockEntries(locked); |
| } |
| } |
| |
| /** |
| * @param entries Entries to update. |
| * @param ver Cache version. |
| * @param writeVals Cache values. |
| * @param putMap Values to put. |
| * @param rmvKeys Keys to remove. |
| * @param expiryPlc Expiry policy. |
| * @param err Optional partial update exception. |
| * @param taskName Task name. |
| * @param transformed {@code True} if transform operation performed. |
| * @return Partial update exception. |
| */ |
| @SuppressWarnings({"unchecked", "ConstantConditions"}) |
| @Nullable private CachePartialUpdateCheckedException updatePartialBatch( |
| List<GridCacheEntryEx> entries, |
| final GridCacheVersion ver, |
| @Nullable List<CacheObject> writeVals, |
| @Nullable Map<KeyCacheObject, CacheObject> putMap, |
| @Nullable Collection<KeyCacheObject> rmvKeys, |
| @Nullable ExpiryPolicy expiryPlc, |
| boolean keepBinary, |
| @Nullable CachePartialUpdateCheckedException err, |
| String taskName, |
| boolean transformed) { |
| assert putMap == null ^ rmvKeys == null; |
| GridCacheOperation op; |
| |
| CacheStorePartialUpdateException storeErr = null; |
| |
| try { |
| if (putMap != null) { |
| try { |
| Map<? extends KeyCacheObject, IgniteBiTuple<? extends CacheObject, GridCacheVersion>> view = F.viewReadOnly(putMap, |
| new C1<CacheObject, IgniteBiTuple<? extends CacheObject, GridCacheVersion>>() { |
| @Override public IgniteBiTuple<? extends CacheObject, GridCacheVersion> apply(CacheObject val) { |
| return F.t(val, ver); |
| } |
| }); |
| |
| ctx.store().putAll(null, view); |
| } |
| catch (CacheStorePartialUpdateException e) { |
| storeErr = e; |
| } |
| |
| op = UPDATE; |
| } |
| else { |
| try { |
| ctx.store().removeAll(null, rmvKeys); |
| } |
| catch (CacheStorePartialUpdateException e) { |
| storeErr = e; |
| } |
| |
| op = DELETE; |
| } |
| } |
| catch (IgniteCheckedException e) { |
| if (err == null) |
| err = partialUpdateException(); |
| |
| err.add(putMap != null ? putMap.keySet() : rmvKeys, e); |
| |
| return err; |
| } |
| |
| boolean intercept = ctx.config().getInterceptor() != null; |
| |
| for (int i = 0; i < entries.size(); i++) { |
| GridCacheEntryEx entry = entries.get(i); |
| |
| assert entry.lockedByCurrentThread(); |
| |
| if (entry.obsolete() || |
| (storeErr != null && storeErr.failedKeys().contains(entry.key().value(ctx.cacheObjectContext(), false)))) |
| continue; |
| |
| try { |
| // We are holding java-level locks on entries at this point. |
| CacheObject writeVal = op == UPDATE ? writeVals.get(i) : null; |
| |
| assert writeVal != null || op == DELETE : "null write value found."; |
| |
| GridTuple3<Boolean, Object, EntryProcessorResult<Object>> t = entry.innerUpdateLocal( |
| ver, |
| op, |
| writeVal, |
| null, |
| false, |
| false, |
| false, |
| keepBinary, |
| expiryPlc, |
| true, |
| true, |
| null, |
| false, |
| taskName, |
| transformed); |
| |
| if (intercept) { |
| if (op == UPDATE) |
| ctx.config().getInterceptor().onAfterPut(new CacheLazyEntry(ctx, entry.key(), writeVal, keepBinary)); |
| else |
| ctx.config().getInterceptor().onAfterRemove(new CacheLazyEntry(ctx, entry.key(), t.get2(), keepBinary)); |
| } |
| } |
| catch (GridCacheEntryRemovedException ignore) { |
| assert false : "Entry cannot become obsolete while holding lock."; |
| } |
| catch (IgniteCheckedException e) { |
| if (err == null) |
| err = partialUpdateException(); |
| |
| err.add(Collections.singleton(entry.key()), e); |
| } |
| } |
| |
| return err; |
| } |
| |
| /** |
| * Acquires java-level locks on cache entries. |
| * |
| * @param keys Keys to lock. |
| * @return Collection of locked entries. |
| */ |
| private List<GridCacheEntryEx> lockEntries(Collection<? extends K> keys) { |
| GridCacheEntryEx[] locked = new GridCacheEntryEx[keys.size()]; |
| |
| boolean nullKeys = false; |
| |
| while (true) { |
| int i = 0; |
| |
| for (K key : keys) { |
| if (key == null) { |
| nullKeys = true; |
| |
| break; |
| } |
| |
| GridCacheEntryEx entry = entryEx(ctx.toCacheKeyObject(key)); |
| |
| locked[i++] = entry; |
| } |
| |
| if (nullKeys) |
| break; |
| |
| if (lockedEntriesInfo.tryLockEntries(locked)) |
| return Arrays.asList(locked); |
| } |
| |
| assert nullKeys; |
| |
| AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion(); |
| |
| for (GridCacheEntryEx entry : locked) { |
| if (entry != null) |
| entry.touch(); |
| } |
| |
| throw new NullPointerException("Null key."); |
| } |
| |
| /** |
| * Releases java-level locks on cache entries. |
| * |
| * @param locked Locked entries. |
| */ |
| private void unlockEntries(Iterable<GridCacheEntryEx> locked) { |
| for (GridCacheEntryEx entry : locked) |
| entry.unlockEntry(); |
| |
| AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion(); |
| |
| for (GridCacheEntryEx entry : locked) |
| entry.touch(); |
| } |
| |
| /** |
| * @return New partial update exception. |
| */ |
| private static CachePartialUpdateCheckedException partialUpdateException() { |
| return new CachePartialUpdateCheckedException("Failed to update keys (retry update if possible)."); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<Boolean> txLockAsync(Collection<KeyCacheObject> keys, |
| long timeout, |
| IgniteTxLocalEx tx, |
| boolean isRead, |
| boolean retval, |
| TransactionIsolation isolation, |
| boolean invalidate, |
| long createTtl, |
| long accessTtl) { |
| return new GridFinishedFuture<>(new UnsupportedOperationException("Locks are not supported for " + |
| "CacheAtomicityMode.ATOMIC mode (use CacheAtomicityMode.TRANSACTIONAL instead)")); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<Boolean> lockAllAsync(@Nullable Collection<? extends K> keys, |
| long timeout) { |
| return new GridFinishedFuture<>(new UnsupportedOperationException("Locks are not supported for " + |
| "CacheAtomicityMode.ATOMIC mode (use CacheAtomicityMode.TRANSACTIONAL instead)")); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void unlockAll(@Nullable Collection<? extends K> keys) throws IgniteCheckedException { |
| throw new UnsupportedOperationException("Locks are not supported for " + |
| "CacheAtomicityMode.ATOMIC mode (use CacheAtomicityMode.TRANSACTIONAL instead)"); |
| } |
| |
| /** |
| * @return Expiry policy. |
| */ |
| @Nullable private ExpiryPolicy expiryPerCall() { |
| CacheOperationContext opCtx = ctx.operationContextPerCall(); |
| |
| ExpiryPolicy expiry = opCtx != null ? opCtx.expiry() : null; |
| |
| if (expiry == null) |
| expiry = ctx.expiry(); |
| |
| return expiry; |
| } |
| |
| /** |
| * @param op Operation closure. |
| * @return Future. |
| */ |
| @SuppressWarnings("unchecked") |
| private IgniteInternalFuture asyncOp(final Callable<?> op) { |
| IgniteInternalFuture fail = asyncOpAcquire(/*retry*/false); |
| |
| if (fail != null) |
| return fail; |
| |
| IgniteInternalFuture f = ctx.closures().callLocalSafe(op); |
| |
| f.listen(new CI1<IgniteInternalFuture<?>>() { |
| @Override public void apply(IgniteInternalFuture<?> f) { |
| asyncOpRelease(false); |
| } |
| }); |
| |
| return f; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onDeferredDelete(GridCacheEntryEx entry, GridCacheVersion ver) { |
| assert false : "Should not be called"; |
| } |
| } |
