| /* |
| * 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.datastructures; |
| |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.Map; |
| import java.util.UUID; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.CountDownLatch; |
| import javax.cache.event.CacheEntryEvent; |
| import javax.cache.event.CacheEntryListenerException; |
| import javax.cache.event.CacheEntryUpdatedListener; |
| import javax.cache.event.EventType; |
| import org.apache.ignite.IgniteAtomicLong; |
| import org.apache.ignite.IgniteAtomicReference; |
| import org.apache.ignite.IgniteAtomicSequence; |
| import org.apache.ignite.IgniteAtomicStamped; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.IgniteCountDownLatch; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.IgniteLock; |
| import org.apache.ignite.IgniteLogger; |
| import org.apache.ignite.IgniteQueue; |
| import org.apache.ignite.IgniteSemaphore; |
| import org.apache.ignite.IgniteSet; |
| import org.apache.ignite.cache.CacheEntryEventSerializableFilter; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.configuration.AtomicConfiguration; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.CollectionConfiguration; |
| import org.apache.ignite.events.DiscoveryEvent; |
| import org.apache.ignite.events.Event; |
| import org.apache.ignite.internal.GridKernalContext; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.IgniteInterruptedCheckedException; |
| import org.apache.ignite.internal.NodeStoppingException; |
| import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; |
| import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException; |
| import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener; |
| import org.apache.ignite.internal.managers.systemview.walker.AtomicLongViewWalker; |
| import org.apache.ignite.internal.managers.systemview.walker.AtomicReferenceViewWalker; |
| import org.apache.ignite.internal.managers.systemview.walker.AtomicSequenceViewWalker; |
| import org.apache.ignite.internal.managers.systemview.walker.AtomicStampedViewWalker; |
| import org.apache.ignite.internal.managers.systemview.walker.CountDownLatchViewWalker; |
| import org.apache.ignite.internal.managers.systemview.walker.QueueViewWalker; |
| import org.apache.ignite.internal.managers.systemview.walker.ReentrantLockViewWalker; |
| import org.apache.ignite.internal.managers.systemview.walker.SemaphoreViewWalker; |
| import org.apache.ignite.internal.managers.systemview.walker.SetViewWalker; |
| import org.apache.ignite.internal.processors.GridProcessorAdapter; |
| import org.apache.ignite.internal.processors.cache.CacheType; |
| import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor; |
| import org.apache.ignite.internal.processors.cache.GridCacheAdapter; |
| import org.apache.ignite.internal.processors.cache.GridCacheContext; |
| import org.apache.ignite.internal.processors.cache.GridCacheInternal; |
| import org.apache.ignite.internal.processors.cache.GridCacheUtils; |
| import org.apache.ignite.internal.processors.cache.IgniteInternalCache; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal; |
| import org.apache.ignite.internal.processors.cluster.IgniteChangeGlobalStateSupport; |
| import org.apache.ignite.internal.util.lang.GridPlainCallable; |
| import org.apache.ignite.internal.util.lang.IgniteClosureX; |
| import org.apache.ignite.internal.util.lang.IgniteInClosureX; |
| import org.apache.ignite.internal.util.lang.IgniteOutClosureX; |
| import org.apache.ignite.internal.util.lang.IgnitePredicateX; |
| import org.apache.ignite.internal.util.lang.gridfunc.PredicateCollectionView; |
| import org.apache.ignite.internal.util.lang.gridfunc.TransformCollectionView; |
| import org.apache.ignite.internal.util.typedef.CI1; |
| import org.apache.ignite.internal.util.typedef.CIX1; |
| 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.X; |
| import org.apache.ignite.internal.util.typedef.internal.A; |
| import org.apache.ignite.internal.util.typedef.internal.GPR; |
| import org.apache.ignite.internal.util.typedef.internal.S; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgnitePredicate; |
| import org.apache.ignite.lang.IgniteProductVersion; |
| import org.apache.ignite.spi.systemview.view.datastructures.AtomicLongView; |
| import org.apache.ignite.spi.systemview.view.datastructures.AtomicReferenceView; |
| import org.apache.ignite.spi.systemview.view.datastructures.AtomicSequenceView; |
| import org.apache.ignite.spi.systemview.view.datastructures.AtomicStampedView; |
| import org.apache.ignite.spi.systemview.view.datastructures.CountDownLatchView; |
| import org.apache.ignite.spi.systemview.view.datastructures.QueueView; |
| import org.apache.ignite.spi.systemview.view.datastructures.ReentrantLockView; |
| import org.apache.ignite.spi.systemview.view.datastructures.SemaphoreView; |
| import org.apache.ignite.spi.systemview.view.datastructures.SetView; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL; |
| import static org.apache.ignite.cache.CacheMode.PARTITIONED; |
| import static org.apache.ignite.cache.CacheRebalanceMode.SYNC; |
| import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; |
| 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.datastructures.DataStructureType.ATOMIC_LONG; |
| import static org.apache.ignite.internal.processors.datastructures.DataStructureType.ATOMIC_REF; |
| import static org.apache.ignite.internal.processors.datastructures.DataStructureType.ATOMIC_SEQ; |
| import static org.apache.ignite.internal.processors.datastructures.DataStructureType.ATOMIC_STAMPED; |
| import static org.apache.ignite.internal.processors.datastructures.DataStructureType.COUNT_DOWN_LATCH; |
| import static org.apache.ignite.internal.processors.datastructures.DataStructureType.QUEUE; |
| import static org.apache.ignite.internal.processors.datastructures.DataStructureType.REENTRANT_LOCK; |
| import static org.apache.ignite.internal.processors.datastructures.DataStructureType.SEMAPHORE; |
| import static org.apache.ignite.internal.processors.datastructures.DataStructureType.SET; |
| import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName; |
| import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC; |
| import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ; |
| |
| /** |
| * Manager of data structures. |
| */ |
| public final class DataStructuresProcessor extends GridProcessorAdapter implements IgniteChangeGlobalStateSupport { |
| /** DataRegionConfiguration name reserved for volatile caches. */ |
| public static final String VOLATILE_DATA_REGION_NAME = "volatileDsMemPlc"; |
| |
| /** */ |
| public static final String DEFAULT_VOLATILE_DS_GROUP_NAME = "default-volatile-ds-group"; |
| |
| /** */ |
| public static final String DEFAULT_DS_GROUP_NAME = "default-ds-group"; |
| |
| /** */ |
| private static final String DS_CACHE_NAME_PREFIX = "datastructures_"; |
| |
| /** Atomics system cache name. */ |
| public static final String ATOMICS_CACHE_NAME = "ignite-sys-atomic-cache"; |
| |
| /** */ |
| public static final String QUEUES_VIEW = metricName("ds", "queues"); |
| |
| /** */ |
| public static final String SETS_VIEW = metricName("ds", "sets"); |
| |
| /** */ |
| public static final String LOCKS_VIEW = metricName("ds", "reentrantlocks"); |
| |
| /** */ |
| public static final String SEMAPHORES_VIEW = metricName("ds", "semaphores"); |
| |
| /** */ |
| public static final String LATCHES_VIEW = metricName("ds", "countdownlatches"); |
| |
| /** */ |
| public static final String STAMPED_VIEW = metricName("ds", "atomicstamped"); |
| |
| /** */ |
| public static final String REFERENCES_VIEW = metricName("ds", "atomicreferences"); |
| |
| /** */ |
| public static final String LONGS_VIEW = metricName("ds", "atomiclongs"); |
| |
| /** */ |
| public static final String SEQUENCES_VIEW = metricName("ds", "atomicsequences"); |
| |
| /** */ |
| private static final String QUEUES_VIEW_DESC = "Data structure queues"; |
| |
| /** */ |
| private static final String SETS_VIEW_DESC = "Data structure sets"; |
| |
| /** */ |
| private static final String LOCKS_VIEW_DESC = "Data structure reentrant locks"; |
| |
| /** */ |
| private static final String SEMAPHORES_VIEW_DESC = "Data structure semaphores"; |
| |
| /** */ |
| private static final String LATCHES_VIEW_DESC = "Data structure count down latches"; |
| |
| /** */ |
| private static final String STAMPED_VIEW_DESC = "Data structure atomic stamped"; |
| |
| /** */ |
| private static final String REFERENCES_VIEW_DESC = "Data structure atomic references"; |
| |
| /** */ |
| private static final String LONGS_VIEW_DESC = "Data structure atomic longs"; |
| |
| /** */ |
| private static final String SEQUENCES_VIEW_DESC = "Data structure atomic sequences"; |
| |
| /** Non collocated IgniteSet will use separate cache if all nodes in cluster is not older then specified version. */ |
| private static final IgniteProductVersion SEPARATE_CACHE_PER_NON_COLLOCATED_SET_SINCE = |
| IgniteProductVersion.fromString("2.7.0"); |
| |
| /** Initial capacity. */ |
| private static final int INITIAL_CAPACITY = 10; |
| |
| /** Initialization latch. */ |
| private volatile CountDownLatch initLatch = new CountDownLatch(1); |
| |
| /** Initialization failed flag. */ |
| private boolean initFailed; |
| |
| /** Internal storage of all dataStructures items (sequence, atomic long etc.). */ |
| private final ConcurrentMap<GridCacheInternalKey, GridCacheRemovable> dsMap; |
| |
| /** Atomic data structures configuration. */ |
| private final AtomicConfiguration dfltAtomicCfg; |
| |
| /** Map of continuous query IDs. */ |
| private final ConcurrentHashMap<Integer, UUID> qryIdMap = new ConcurrentHashMap<>(); |
| |
| /** Listener. */ |
| private final GridLocalEventListener lsnr = new GridLocalEventListener() { |
| @Override public void onEvent(final Event evt) { |
| // This may require cache operation to execute, |
| // therefore cannot use event notification thread. |
| ctx.closure().callLocalSafe( |
| new GridPlainCallable<Object>() { |
| @Override public Object call() { |
| DiscoveryEvent discoEvt = (DiscoveryEvent)evt; |
| |
| UUID leftNodeId = discoEvt.eventNode().id(); |
| |
| for (GridCacheRemovable ds : dsMap.values()) { |
| if (ds instanceof GridCacheSemaphoreEx) |
| ((GridCacheSemaphoreEx)ds).onNodeRemoved(leftNodeId); |
| else if (ds instanceof GridCacheLockEx) |
| ((GridCacheLockEx)ds).onNodeRemoved(leftNodeId); |
| } |
| |
| return null; |
| } |
| }, |
| false); |
| } |
| }; |
| |
| /** |
| * @param ctx Context. |
| */ |
| public DataStructuresProcessor(GridKernalContext ctx) { |
| super(ctx); |
| |
| dsMap = new ConcurrentHashMap<>(INITIAL_CAPACITY); |
| |
| dfltAtomicCfg = ctx.config().getAtomicConfiguration(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void start() { |
| ctx.event().addLocalEventListener(lsnr, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onKernalStart(boolean active) { |
| registerSystemViews(); |
| |
| if (!active) |
| return; |
| |
| onKernalStart0(); |
| } |
| |
| /** |
| * |
| */ |
| public void onBeforeActivate() { |
| CountDownLatch latch0 = initLatch; |
| |
| if (latch0 == null || latch0.getCount() == 0) |
| initLatch = new CountDownLatch(1); |
| } |
| |
| /** |
| * |
| */ |
| private void onKernalStart0() { |
| initLatch.countDown(); |
| } |
| |
| /** |
| * @param cctx Cache context. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private void startQuery(GridCacheContext cctx) throws IgniteCheckedException { |
| if (!qryIdMap.containsKey(cctx.cacheId())) { |
| synchronized (this) { |
| if (!qryIdMap.containsKey(cctx.cacheId())) { |
| qryIdMap.put(cctx.cacheId(), |
| cctx.continuousQueries().executeInternalQuery( |
| new DataStructuresEntryListener(), |
| new DataStructuresEntryFilter(), |
| cctx.isReplicated() && cctx.affinityNode(), |
| false, |
| false, |
| true |
| )); |
| } |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onKernalStop(boolean cancel) { |
| super.onKernalStop(cancel); |
| |
| for (GridCacheRemovable ds : dsMap.values()) { |
| if (ds instanceof GridCacheSemaphoreEx) |
| ((GridCacheSemaphoreEx)ds).stop(); |
| |
| if (ds instanceof GridCacheLockEx) |
| ((GridCacheLockEx)ds).onStop(); |
| } |
| |
| CountDownLatch init0 = initLatch; |
| |
| if (init0 != null && init0.getCount() > 0) { |
| initFailed = true; |
| |
| init0.countDown(); |
| |
| initLatch = null; |
| } |
| |
| Iterator<Map.Entry<Integer, UUID>> iter = qryIdMap.entrySet().iterator(); |
| |
| while (iter.hasNext()) { |
| Map.Entry<Integer, UUID> e = iter.next(); |
| |
| iter.remove(); |
| |
| GridCacheContext cctx = ctx.cache().context().cacheContext(e.getKey()); |
| |
| cctx.continuousQueries().cancelInternalQuery(e.getValue()); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onActivate(GridKernalContext ctx) { |
| if (log.isDebugEnabled()) |
| log.debug("Activating data structure processor [nodeId=" + ctx.localNodeId() + |
| " topVer=" + ctx.discovery().topologyVersionEx() + " ]"); |
| |
| initFailed = false; |
| |
| qryIdMap.clear(); |
| |
| ctx.event().addLocalEventListener(lsnr, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| |
| restoreStructuresState(ctx); |
| |
| onKernalStart0(); |
| } |
| |
| /** Register system views. */ |
| private void registerSystemViews() { |
| ctx.systemView().registerView( |
| SEQUENCES_VIEW, |
| SEQUENCES_VIEW_DESC, |
| new AtomicSequenceViewWalker(), |
| new PredicateCollectionView<>(dsMap.values(), v -> v instanceof IgniteAtomicSequence), |
| AtomicSequenceView::new |
| ); |
| |
| ctx.systemView().registerView( |
| LONGS_VIEW, |
| LONGS_VIEW_DESC, |
| new AtomicLongViewWalker(), |
| new PredicateCollectionView<>(dsMap.values(), v -> v instanceof IgniteAtomicLong), |
| AtomicLongView::new |
| ); |
| |
| ctx.systemView().registerView( |
| REFERENCES_VIEW, |
| REFERENCES_VIEW_DESC, |
| new AtomicReferenceViewWalker(), |
| new PredicateCollectionView<>(dsMap.values(), v -> v instanceof IgniteAtomicReference), |
| AtomicReferenceView::new |
| ); |
| |
| ctx.systemView().registerView( |
| STAMPED_VIEW, |
| STAMPED_VIEW_DESC, |
| new AtomicStampedViewWalker(), |
| new PredicateCollectionView<>(dsMap.values(), v -> v instanceof IgniteAtomicStamped), |
| AtomicStampedView::new |
| ); |
| |
| ctx.systemView().registerView( |
| LATCHES_VIEW, |
| LATCHES_VIEW_DESC, |
| new CountDownLatchViewWalker(), |
| new PredicateCollectionView<>(dsMap.values(), v -> v instanceof IgniteCountDownLatch), |
| CountDownLatchView::new |
| ); |
| |
| ctx.systemView().registerView( |
| SEMAPHORES_VIEW, |
| SEMAPHORES_VIEW_DESC, |
| new SemaphoreViewWalker(), |
| new PredicateCollectionView<>(dsMap.values(), v -> v instanceof IgniteSemaphore), |
| SemaphoreView::new |
| ); |
| |
| ctx.systemView().registerView( |
| LOCKS_VIEW, |
| LOCKS_VIEW_DESC, |
| new ReentrantLockViewWalker(), |
| new PredicateCollectionView<>(dsMap.values(), v -> v instanceof IgniteLock), |
| ReentrantLockView::new |
| ); |
| |
| ctx.systemView().registerInnerCollectionView( |
| QUEUES_VIEW, |
| QUEUES_VIEW_DESC, |
| new QueueViewWalker(), |
| new TransformCollectionView<>( |
| ctx.cache().cacheDescriptors().values(), |
| desc -> ctx.cache().cache(desc.cacheName()).context().dataStructures(), |
| desc -> desc.cacheType() == CacheType.DATA_STRUCTURES), |
| cctx -> cctx.queues().values(), |
| (cctx, queue) -> new QueueView(queue) |
| ); |
| |
| ctx.systemView().registerInnerCollectionView( |
| SETS_VIEW, |
| SETS_VIEW_DESC, |
| new SetViewWalker(), |
| F.viewReadOnly( |
| ctx.cache().cacheDescriptors().values(), |
| desc -> ctx.cache().cache(desc.cacheName()).context().dataStructures(), |
| desc -> desc.cacheType() == CacheType.DATA_STRUCTURES), |
| cctx -> cctx.sets().values(), |
| (cctx, set) -> new SetView(set) |
| ); |
| } |
| |
| /** |
| * @param ctx Context. |
| */ |
| public void restoreStructuresState(GridKernalContext ctx) { |
| onKernalStart0(); |
| |
| try { |
| for (GridCacheRemovable v : dsMap.values()) { |
| if (v instanceof IgniteChangeGlobalStateSupport) |
| ((IgniteChangeGlobalStateSupport)v).onActivate(ctx); |
| } |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed restore data structures state", e); |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onDeActivate(GridKernalContext ctx) { |
| if (log.isDebugEnabled()) |
| log.debug("DeActivate data structure processor [nodeId=" + ctx.localNodeId() + |
| ", topVer=" + ctx.discovery().topologyVersionEx() + "]"); |
| |
| ctx.event().removeLocalEventListener(lsnr, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| |
| onKernalStop(false); |
| |
| initLatch = null; |
| |
| for (GridCacheRemovable v : dsMap.values()) { |
| if (v instanceof IgniteChangeGlobalStateSupport) |
| ((IgniteChangeGlobalStateSupport)v).onDeActivate(ctx); |
| } |
| } |
| |
| /** |
| * @param key Key. |
| * @param obj Object. |
| */ |
| void onRemoved(GridCacheInternalKey key, GridCacheRemovable obj) { |
| dsMap.remove(key, obj); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public IgniteInternalFuture<?> onReconnected(boolean clusterRestarted) throws IgniteCheckedException { |
| UUID nodeId = ctx.localNodeId(); |
| |
| for (Map.Entry<GridCacheInternalKey, GridCacheRemovable> e : dsMap.entrySet()) { |
| GridCacheRemovable obj = e.getValue(); |
| |
| if (clusterRestarted) { |
| obj.onRemoved(); |
| |
| dsMap.remove(e.getKey(), obj); |
| } |
| else |
| obj.needCheckNotRemoved(); |
| |
| if (obj instanceof GridCacheLockEx) |
| ((GridCacheLockEx)obj).onReconnected(nodeId); |
| } |
| |
| for (GridCacheContext cctx : ctx.cache().context().cacheContexts()) |
| cctx.dataStructures().onReconnected(clusterRestarted); |
| |
| return null; |
| } |
| |
| /** |
| * @param cacheName Cache name. |
| * @return {@code True} if cache with such name is used to store data structures. |
| */ |
| public static boolean isDataStructureCache(String cacheName) { |
| return cacheName != null && (cacheName.startsWith(ATOMICS_CACHE_NAME) || |
| cacheName.startsWith(DS_CACHE_NAME_PREFIX) || |
| cacheName.equals(DEFAULT_DS_GROUP_NAME) || |
| cacheName.equals(DEFAULT_VOLATILE_DS_GROUP_NAME)); |
| } |
| |
| /** |
| * @param grpName Group name. |
| * @return {@code True} if group name is reserved to store data structures. |
| */ |
| public static boolean isReservedGroup(@Nullable String grpName) { |
| return grpName != null && |
| (DEFAULT_DS_GROUP_NAME.equals(grpName) || |
| grpName.startsWith(DEFAULT_VOLATILE_DS_GROUP_NAME)); |
| } |
| |
| /** |
| * Gets a sequence from cache or creates one if it's not cached. |
| * |
| * @param name Sequence name. |
| * @param cfg Configuration. |
| * @param initVal Initial value for sequence. If sequence already cached, {@code initVal} will be ignored. |
| * @param create If {@code true} sequence will be created in case it is not in cache. |
| * @return Sequence. |
| * @throws IgniteCheckedException If loading failed. |
| */ |
| public final IgniteAtomicSequence sequence(final String name, |
| @Nullable final AtomicConfiguration cfg, |
| final long initVal, |
| final boolean create |
| ) throws IgniteCheckedException { |
| return getAtomic(new AtomicAccessor<GridCacheAtomicSequenceEx>() { |
| @Override public T2<GridCacheAtomicSequenceEx, AtomicDataStructureValue> get(GridCacheInternalKey key, |
| AtomicDataStructureValue val, IgniteInternalCache cache) throws IgniteCheckedException { |
| GridCacheAtomicSequenceValue seqVal = cast(val, GridCacheAtomicSequenceValue.class); |
| |
| // Check that sequence hasn't been created in other thread yet. |
| GridCacheAtomicSequenceEx seq = cast(dsMap.get(key), GridCacheAtomicSequenceEx.class); |
| |
| if (seq != null) { |
| assert seqVal != null; |
| |
| return new T2<>(seq, null); |
| } |
| |
| if (seqVal == null && !create) |
| return null; |
| |
| AtomicConfiguration cfg0 = cfg != null ? cfg : dfltAtomicCfg; |
| |
| // We should use offset because we already reserved left side of range. |
| long off = cfg0.getAtomicSequenceReserveSize() > 1 ? cfg0.getAtomicSequenceReserveSize() - 1 : 1; |
| |
| long upBound; |
| long locCntr; |
| |
| if (seqVal == null) { |
| locCntr = initVal; |
| |
| upBound = locCntr + off; |
| |
| // Global counter must be more than reserved region. |
| seqVal = new GridCacheAtomicSequenceValue(upBound + 1); |
| } |
| else { |
| locCntr = seqVal.get(); |
| |
| upBound = locCntr + off; |
| |
| // Global counter must be more than reserved region. |
| seqVal.set(upBound + 1); |
| } |
| |
| // Only one thread can be in the transaction scope and create sequence. |
| seq = new GridCacheAtomicSequenceImpl(name, |
| key, |
| cache, |
| cfg0.getAtomicSequenceReserveSize(), |
| locCntr, |
| upBound); |
| |
| return new T2<GridCacheAtomicSequenceEx, AtomicDataStructureValue>(seq, seqVal); |
| } |
| }, cfg, name, DataStructureType.ATOMIC_SEQ, create, GridCacheAtomicSequenceEx.class); |
| } |
| |
| /** |
| * Removes sequence from cache. |
| * |
| * @param name Sequence name. |
| * @param grpName Group name. |
| * @throws IgniteCheckedException If removing failed. |
| */ |
| final void removeSequence(final String name, final String grpName) throws IgniteCheckedException { |
| removeDataStructure(null, name, grpName, ATOMIC_SEQ, null); |
| } |
| |
| /** |
| * Gets an atomic long from cache or creates one if it's not cached. |
| * |
| * @param name Name of atomic long. |
| * @param cfg Configuration. |
| * @param initVal Initial value for atomic long. If atomic long already cached, {@code initVal} |
| * will be ignored. |
| * @param create If {@code true} atomic long will be created in case it is not in cache. |
| * @return Atomic long. |
| * @throws IgniteCheckedException If loading failed. |
| */ |
| public final IgniteAtomicLong atomicLong(final String name, |
| @Nullable AtomicConfiguration cfg, |
| final long initVal, |
| final boolean create) throws IgniteCheckedException { |
| return getAtomic(new AtomicAccessor<GridCacheAtomicLongEx>() { |
| @Override public T2<GridCacheAtomicLongEx, AtomicDataStructureValue> get( |
| GridCacheInternalKey key, |
| AtomicDataStructureValue val, |
| IgniteInternalCache cache |
| ) throws IgniteCheckedException { |
| // Check that atomic long hasn't been created in other thread yet. |
| GridCacheAtomicLongEx a = cast(dsMap.get(key), GridCacheAtomicLongEx.class); |
| |
| if (a != null) { |
| assert val != null; |
| |
| return new T2<>(a, null); |
| } |
| |
| if (val == null && !create) |
| return null; |
| |
| GridCacheAtomicLongValue retVal = (val == null ? new GridCacheAtomicLongValue(initVal) : null); |
| |
| a = new GridCacheAtomicLongImpl(name, key, cache); |
| |
| return new T2<GridCacheAtomicLongEx, AtomicDataStructureValue>(a, retVal); |
| } |
| }, cfg, name, ATOMIC_LONG, create, GridCacheAtomicLongEx.class); |
| } |
| |
| /** |
| * @param c Closure creating data structure instance. |
| * @param cfg Optional custom configuration or {@code null} to use default one. |
| * @param name Data structure name. |
| * @param type Data structure type. |
| * @param create Create flag. |
| * @param cls Expected data structure class. |
| * @return Data structure instance. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @Nullable private <T extends GridCacheRemovable> T getAtomic(final AtomicAccessor<T> c, |
| @Nullable AtomicConfiguration cfg, |
| final String name, |
| final DataStructureType type, |
| final boolean create, |
| Class<? extends T> cls |
| ) throws IgniteCheckedException { |
| A.notNull(name, "name"); |
| |
| awaitInitialization(); |
| |
| if (cfg == null) { |
| checkAtomicsConfiguration(); |
| |
| cfg = dfltAtomicCfg; |
| } |
| |
| String dataRegionName = null; |
| final String grpName; |
| |
| if (type.isVolatile()) { |
| String volatileGrpName = DEFAULT_VOLATILE_DS_GROUP_NAME; |
| |
| dataRegionName = VOLATILE_DATA_REGION_NAME; |
| |
| volatileGrpName += "@" + dataRegionName; |
| |
| grpName = volatileGrpName; |
| } |
| else if (cfg.getGroupName() != null) |
| grpName = cfg.getGroupName(); |
| else |
| grpName = DEFAULT_DS_GROUP_NAME; |
| |
| String cacheName = ATOMICS_CACHE_NAME + "@" + grpName; |
| |
| IgniteInternalCache<GridCacheInternalKey, AtomicDataStructureValue> cache0 = ctx.cache().cache(cacheName); |
| |
| if (cache0 == null) { |
| if (!create && ctx.cache().cacheDescriptor(cacheName) == null) |
| return null; |
| |
| ctx.cache().dynamicStartCache(cacheConfiguration(cfg, cacheName, grpName, dataRegionName), |
| cacheName, |
| null, |
| CacheType.DATA_STRUCTURES, |
| false, |
| false, |
| true, |
| true).get(); |
| |
| cache0 = ctx.cache().cache(cacheName); |
| |
| assert cache0 != null; |
| } |
| |
| final IgniteInternalCache<GridCacheInternalKey, AtomicDataStructureValue> cache = cache0; |
| |
| startQuery(cache.context()); |
| |
| final GridCacheInternalKey key = new GridCacheInternalKeyImpl(name, grpName); |
| |
| // Check type of structure received by key from local cache. |
| T dataStructure = cast(dsMap.get(key), cls); |
| |
| if (dataStructure != null) |
| return dataStructure; |
| |
| return retryTopologySafe(new IgniteOutClosureX<T>() { |
| @Override public T applyx() throws IgniteCheckedException { |
| cache.context().gate().enter(); |
| |
| try (GridNearTxLocal tx = cache.txStartEx(PESSIMISTIC, REPEATABLE_READ)) { |
| AtomicDataStructureValue val = cache.get(key); |
| |
| if (isObsolete(val)) |
| val = null; |
| |
| if (val == null && !create) |
| return null; |
| |
| if (val != null) { |
| if (val.type() != type) |
| throw new IgniteCheckedException("Another data structure with the same name already created " + |
| "[name=" + name + |
| ", newType=" + type + |
| ", existingType=" + val.type() + ']'); |
| } |
| |
| T2<T, ? extends AtomicDataStructureValue> ret; |
| |
| try { |
| ret = c.get(key, val, cache); |
| |
| dsMap.put(key, ret.get1()); |
| |
| if (ret.get2() != null) |
| cache.put(key, ret.get2()); |
| |
| tx.commit(); |
| } |
| catch (Error | Exception e) { |
| dsMap.remove(key); |
| |
| U.error(log, "Failed to make datastructure: " + name, e); |
| |
| throw e; |
| } |
| |
| return ret.get1(); |
| } |
| finally { |
| cache.context().gate().leave(); |
| } |
| } |
| }); |
| } |
| |
| /** |
| * @param val Value. |
| * @return {@code True} if value is obsolete. |
| */ |
| private boolean isObsolete(AtomicDataStructureValue val) { |
| return !(val == null || !(val instanceof VolatileAtomicDataStructureValue)) && |
| ((VolatileAtomicDataStructureValue)val).gridStartTime() != ctx.discovery().gridStartTime(); |
| |
| } |
| |
| /** |
| * Removes atomic long from cache. |
| * |
| * @param name Atomic long name. |
| * @param grpName Group name. |
| * @throws IgniteCheckedException If removing failed. |
| */ |
| final void removeAtomicLong(final String name, @Nullable final String grpName) throws IgniteCheckedException { |
| removeDataStructure(null, name, grpName, ATOMIC_LONG, null); |
| } |
| |
| /** |
| * @param pred Remove predicate. |
| * @param name Data structure name. |
| * @param grpName Group name. |
| * @param type Data structure type. |
| * @param afterRmv Optional closure to run after data structure removed. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private <T> void removeDataStructure( |
| @Nullable final IgnitePredicateX<AtomicDataStructureValue> pred, |
| final String name, |
| String grpName, |
| final DataStructureType type, |
| @Nullable final IgniteInClosureX<T> afterRmv |
| ) throws IgniteCheckedException { |
| assert name != null; |
| assert grpName != null; |
| assert type != null; |
| |
| awaitInitialization(); |
| |
| final String cacheName = ATOMICS_CACHE_NAME + "@" + grpName; |
| |
| final GridCacheInternalKey key = new GridCacheInternalKeyImpl(name, grpName); |
| |
| retryTopologySafe(new IgniteOutClosureX<Object>() { |
| @Override public Object applyx() throws IgniteCheckedException { |
| IgniteInternalCache<GridCacheInternalKey, AtomicDataStructureValue> cache = ctx.cache().cache(cacheName); |
| |
| if (cache != null && cache.context().gate().enterIfNotStopped()) { |
| boolean isInterrupted = Thread.interrupted(); |
| |
| try { |
| while (true) { |
| try { |
| try (GridNearTxLocal tx = cache.txStartEx(PESSIMISTIC, REPEATABLE_READ)) { |
| AtomicDataStructureValue val = cache.get(key); |
| |
| if (val == null) |
| return null; |
| |
| if (val.type() != type) |
| throw new IgniteCheckedException("Data structure has different type " + |
| "[name=" + name + |
| ", expectedType=" + type + |
| ", actualType=" + val.type() + ']'); |
| |
| if (pred == null || pred.applyx(val)) { |
| cache.remove(key); |
| |
| tx.commit(); |
| |
| if (afterRmv != null) |
| afterRmv.applyx(null); |
| } |
| } |
| |
| break; |
| } |
| catch (IgniteCheckedException e) { |
| if (X.hasCause(e, IgniteInterruptedCheckedException.class, InterruptedException.class)) |
| isInterrupted = Thread.interrupted(); |
| else |
| throw e; |
| } |
| } |
| } |
| finally { |
| cache.context().gate().leave(); |
| |
| if (isInterrupted) |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| return null; |
| } |
| }); |
| } |
| |
| /** |
| * Would suspend calls for this cache if it is atomics cache. |
| * @param cacheName To suspend. |
| */ |
| public void suspend(String cacheName) { |
| for (Map.Entry<GridCacheInternalKey, GridCacheRemovable> e : dsMap.entrySet()) { |
| String cacheName0 = ATOMICS_CACHE_NAME + "@" + e.getKey().groupName(); |
| |
| if (cacheName0.equals(cacheName)) |
| e.getValue().suspend(); |
| } |
| } |
| |
| /** |
| * Would return this cache to normal work if it was suspened (and if it is atomics cache). |
| * @param cacheName To restart. |
| */ |
| public void restart(String cacheName, IgniteInternalCache cache) { |
| for (Map.Entry<GridCacheInternalKey, GridCacheRemovable> e : dsMap.entrySet()) { |
| String cacheName0 = ATOMICS_CACHE_NAME + "@" + e.getKey().groupName(); |
| |
| if (cacheName0.equals(cacheName)) { |
| if (cache != null) |
| e.getValue().restart(cache); |
| else { |
| e.getValue().onRemoved(); |
| |
| dsMap.remove(e.getKey(), e.getValue()); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Gets an atomic reference from cache or creates one if it's not cached. |
| * |
| * @param name Name of atomic reference. |
| * @param cfg Configuration. |
| * @param initVal Initial value for atomic reference. If atomic reference already cached, {@code initVal} |
| * will be ignored. |
| * @param create If {@code true} atomic reference will be created in case it is not in cache. |
| * @return Atomic reference. |
| * @throws IgniteCheckedException If loading failed. |
| */ |
| @SuppressWarnings("unchecked") |
| public final <T> IgniteAtomicReference<T> atomicReference(final String name, |
| @Nullable AtomicConfiguration cfg, |
| final T initVal, |
| final boolean create |
| ) throws IgniteCheckedException { |
| return getAtomic(new AtomicAccessor<GridCacheAtomicReferenceEx>() { |
| @Override public T2<GridCacheAtomicReferenceEx, AtomicDataStructureValue> get( |
| GridCacheInternalKey key, |
| AtomicDataStructureValue val, |
| IgniteInternalCache cache |
| ) throws IgniteCheckedException { |
| // Check that atomic reference hasn't been created in other thread yet. |
| GridCacheAtomicReferenceEx ref = cast(dsMap.get(key), |
| GridCacheAtomicReferenceEx.class); |
| |
| if (ref != null) { |
| assert val != null; |
| |
| return new T2<>(ref, null); |
| } |
| |
| if (val == null && !create) |
| return null; |
| |
| AtomicDataStructureValue retVal = (val == null ? new GridCacheAtomicReferenceValue<>(initVal) : null); |
| |
| ref = new GridCacheAtomicReferenceImpl(name, key, cache); |
| |
| return new T2<>(ref, retVal); |
| } |
| }, cfg, name, ATOMIC_REF, create, GridCacheAtomicReferenceEx.class); |
| } |
| |
| /** |
| * Removes atomic reference from cache. |
| * |
| * @param name Atomic reference name. |
| * @param grpName Group name. |
| * @throws IgniteCheckedException If removing failed. |
| */ |
| final void removeAtomicReference(final String name, @Nullable final String grpName) throws IgniteCheckedException { |
| removeDataStructure(null, name, grpName, ATOMIC_REF, null); |
| } |
| |
| /** |
| * Gets an atomic stamped from cache or creates one if it's not cached. |
| * |
| * @param name Name of atomic stamped. |
| * @param cfg Configuration. |
| * @param initVal Initial value for atomic stamped. If atomic stamped already cached, {@code initVal} |
| * will be ignored. |
| * @param initStamp Initial stamp for atomic stamped. If atomic stamped already cached, {@code initStamp} |
| * will be ignored. |
| * @param create If {@code true} atomic stamped will be created in case it is not in cache. |
| * @return Atomic stamped. |
| * @throws IgniteCheckedException If loading failed. |
| */ |
| @SuppressWarnings("unchecked") |
| public final <T, S> IgniteAtomicStamped<T, S> atomicStamped(final String name, @Nullable AtomicConfiguration cfg, |
| final T initVal, final S initStamp, final boolean create) throws IgniteCheckedException { |
| return getAtomic(new AtomicAccessor<GridCacheAtomicStampedEx>() { |
| @Override public T2<GridCacheAtomicStampedEx, AtomicDataStructureValue> get( |
| GridCacheInternalKey key, |
| AtomicDataStructureValue val, |
| IgniteInternalCache cache |
| ) throws IgniteCheckedException { |
| // Check that atomic stamped hasn't been created in other thread yet. |
| GridCacheAtomicStampedEx stmp = cast(dsMap.get(key), |
| GridCacheAtomicStampedEx.class); |
| |
| if (stmp != null) { |
| assert val != null; |
| |
| return new T2(stmp, null); |
| } |
| |
| if (val == null && !create) |
| return null; |
| |
| AtomicDataStructureValue retVal = (val == null ? new GridCacheAtomicStampedValue(initVal, initStamp) : null); |
| |
| stmp = new GridCacheAtomicStampedImpl(name, key, cache); |
| |
| return new T2<>(stmp, retVal); |
| } |
| }, cfg, name, ATOMIC_STAMPED, create, GridCacheAtomicStampedEx.class); |
| } |
| |
| /** |
| * Removes atomic stamped from cache. |
| * |
| * @param name Atomic stamped name. |
| * @param grpName Group name. |
| * @throws IgniteCheckedException If removing failed. |
| */ |
| final void removeAtomicStamped(final String name, final String grpName) throws IgniteCheckedException { |
| removeDataStructure(null, name, grpName, ATOMIC_STAMPED, null); |
| } |
| |
| /** |
| * Gets a queue from cache or creates one if it's not cached. |
| * |
| * @param name Name of queue. |
| * @param grpName Group name. If present, will override groupName from configuration. |
| * @param cap Max size of queue. |
| * @param cfg Non-null queue configuration if new queue should be created. |
| * @return Instance of queue. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public final <T> IgniteQueue<T> queue(final String name, @Nullable final String grpName, int cap, |
| @Nullable final CollectionConfiguration cfg) |
| throws IgniteCheckedException { |
| A.notNull(name, "name"); |
| |
| if (cfg != null) { |
| if (cap <= 0) |
| cap = Integer.MAX_VALUE; |
| } |
| |
| final int cap0 = cap; |
| |
| final boolean create = cfg != null; |
| |
| return getCollection(new IgniteClosureX<GridCacheContext, IgniteQueue<T>>() { |
| @Override public IgniteQueue<T> applyx(GridCacheContext ctx) throws IgniteCheckedException { |
| return ctx.dataStructures().queue(name, cap0, isCollocated(cfg), create); |
| } |
| }, cfg, name, grpName, QUEUE, create, false); |
| } |
| |
| /** |
| * Non-collocated mode only makes sense for and is only supported for PARTITIONED caches, so |
| * collocated mode should be enabled for non-partitioned cache by default. |
| * |
| * @param cfg Collection configuration. |
| * @return {@code True} If collocated mode should be enabled. |
| */ |
| private boolean isCollocated(CollectionConfiguration cfg) { |
| return cfg != null && (cfg.isCollocated() || cfg.getCacheMode() != PARTITIONED); |
| } |
| |
| /** |
| * @param cfg Atomic configuration. |
| * @param name Cache name. |
| * @param grpName Group name. |
| * @param dataRegionName Name of data region for this cache. |
| * |
| * @return Cache configuration. |
| */ |
| private CacheConfiguration cacheConfiguration(AtomicConfiguration cfg, String name, String grpName, |
| String dataRegionName) { |
| CacheConfiguration ccfg = new CacheConfiguration(); |
| |
| ccfg.setName(name); |
| ccfg.setGroupName(grpName); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setRebalanceMode(SYNC); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setCacheMode(cfg.getCacheMode()); |
| ccfg.setNodeFilter(CacheConfiguration.ALL_NODES); |
| ccfg.setAffinity(cfg.getAffinity()); |
| ccfg.setDataRegionName(dataRegionName); |
| |
| if (cfg.getCacheMode() == PARTITIONED) |
| ccfg.setBackups(cfg.getBackups()); |
| |
| return ccfg; |
| } |
| |
| /** |
| * @param cfg Collection configuration. |
| * @param name Cache name. |
| * @param grpName Group name. |
| * @return Cache configuration. |
| */ |
| private CacheConfiguration cacheConfiguration(CollectionConfiguration cfg, String name, String grpName) { |
| CacheConfiguration ccfg = new CacheConfiguration(); |
| |
| ccfg.setName(name); |
| ccfg.setGroupName(grpName); |
| ccfg.setBackups(cfg.getBackups()); |
| ccfg.setCacheMode(cfg.getCacheMode()); |
| ccfg.setAtomicityMode(cfg.getAtomicityMode()); |
| ccfg.setNodeFilter(cfg.getNodeFilter()); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| return ccfg; |
| } |
| |
| /** |
| * @param cfg Collection configuration. |
| * @param name Cache name. |
| * @param grpName Group name. |
| * @return Meta cache configuration. |
| */ |
| private CacheConfiguration metaCacheConfiguration(CollectionConfiguration cfg, String name, String grpName) { |
| CacheConfiguration ccfg = cacheConfiguration(cfg, name, grpName); |
| |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| |
| return ccfg; |
| } |
| |
| /** |
| * Get compatible with collection configuration data structure cache. |
| * |
| * @param cfg Collection configuration. |
| * @param grpName Group name. |
| * @param dsType Data structure type. |
| * @param dsName Data structure name. |
| * @param separated Separated cache flag. |
| * @return Data structure cache. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private IgniteInternalCache compatibleCache(CollectionConfiguration cfg, |
| String grpName, |
| DataStructureType dsType, |
| String dsName, |
| boolean separated |
| ) throws IgniteCheckedException { |
| String cacheName = DS_CACHE_NAME_PREFIX + cfg.getAtomicityMode() + "_" + cfg.getCacheMode() + "_" + |
| cfg.getBackups() + "@" + grpName; |
| |
| IgniteInternalCache cache = ctx.cache().cache(cacheName); |
| |
| if (separated && (cache == null || !cache.containsKey(new GridCacheSetHeaderKey(dsName)))) { |
| cacheName += "#" + dsType.name() + "_" + dsName; |
| |
| cache = ctx.cache().cache(cacheName); |
| } |
| |
| if (cache == null) { |
| ctx.cache().dynamicStartCache(cacheConfiguration(cfg, cacheName, grpName), |
| cacheName, |
| null, |
| CacheType.DATA_STRUCTURES, |
| false, |
| false, |
| true, |
| true).get(); |
| } |
| else { |
| IgnitePredicate<ClusterNode> cacheNodeFilter = cache.context().group().nodeFilter(); |
| |
| String clsName1 = cacheNodeFilter != null ? cacheNodeFilter.getClass().getName() : |
| CacheConfiguration.IgniteAllNodesPredicate.class.getName(); |
| String clsName2 = cfg.getNodeFilter() != null ? cfg.getNodeFilter().getClass().getName() : |
| CacheConfiguration.IgniteAllNodesPredicate.class.getName(); |
| |
| if (!clsName1.equals(clsName2)) |
| throw new IgniteCheckedException("Could not add collection to group " + grpName + |
| " because of different node filters [existing=" + clsName1 + ", new=" + clsName2 + "]"); |
| } |
| |
| cache = ctx.cache().getOrStartCache(cacheName); |
| |
| assert cache != null; |
| |
| return cache; |
| } |
| |
| /** |
| * @param name Queue name. |
| * @param cctx Queue cache context. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public void removeQueue(final String name, final GridCacheContext cctx) throws IgniteCheckedException { |
| assert name != null; |
| assert cctx != null; |
| |
| CIX1<GridCacheQueueHeader> afterRmv = new CIX1<GridCacheQueueHeader>() { |
| @Override public void applyx(GridCacheQueueHeader hdr) throws IgniteCheckedException { |
| hdr = (GridCacheQueueHeader)cctx.cache().withNoRetries().getAndRemove(new GridCacheQueueHeaderKey(name)); |
| |
| if (hdr == null || hdr.empty()) |
| return; |
| |
| GridCacheQueueAdapter.removeKeys(cctx.cache(), |
| hdr.id(), |
| name, |
| hdr.collocated(), |
| hdr.head(), |
| hdr.tail(), |
| 0); |
| } |
| }; |
| |
| removeDataStructure(null, name, cctx.group().name(), QUEUE, afterRmv); |
| } |
| |
| /** |
| * @param c Closure creating collection. |
| * @param cfg Configuration. |
| * @param name Collection name. |
| * @param grpName Cache group name. |
| * @param type Data structure type. |
| * @param create Create flag. |
| * @param separated Separated cache flag. |
| * @return Collection instance. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @Nullable private <T> T getCollection(final IgniteClosureX<GridCacheContext, T> c, |
| @Nullable CollectionConfiguration cfg, |
| String name, |
| @Nullable String grpName, |
| final DataStructureType type, |
| boolean create, |
| boolean separated |
| ) throws IgniteCheckedException { |
| awaitInitialization(); |
| |
| assert name != null; |
| assert type.isCollection() : type; |
| assert !create || cfg != null; |
| |
| if (grpName == null) { |
| if (cfg != null && cfg.getGroupName() != null) |
| grpName = cfg.getGroupName(); |
| else |
| grpName = DEFAULT_DS_GROUP_NAME; |
| } |
| |
| final String metaCacheName = ATOMICS_CACHE_NAME + "@" + grpName; |
| |
| IgniteInternalCache<GridCacheInternalKey, AtomicDataStructureValue> metaCache0 = ctx.cache().cache(metaCacheName); |
| |
| if (metaCache0 == null) { |
| CacheConfiguration ccfg = null; |
| |
| if (!create) { |
| DynamicCacheDescriptor desc = ctx.cache().cacheDescriptor(metaCacheName); |
| |
| if (desc == null) |
| return null; |
| } |
| else |
| ccfg = metaCacheConfiguration(cfg, metaCacheName, grpName); |
| |
| ctx.cache().dynamicStartCache(ccfg, |
| metaCacheName, |
| null, |
| CacheType.DATA_STRUCTURES, |
| false, |
| false, |
| true, |
| true).get(); |
| |
| metaCache0 = ctx.cache().cache(metaCacheName); |
| |
| assert metaCache0 != null; |
| } |
| |
| final IgniteInternalCache<GridCacheInternalKey, AtomicDataStructureValue> metaCache = metaCache0; |
| |
| AtomicDataStructureValue oldVal; |
| |
| final IgniteInternalCache cache; |
| |
| if (create) { |
| cache = compatibleCache(cfg, grpName, type, name, separated); |
| |
| DistributedCollectionMetadata newVal = new DistributedCollectionMetadata(type, cfg, cache.name()); |
| |
| oldVal = metaCache.getAndPutIfAbsent(new GridCacheInternalKeyImpl(name, grpName), newVal); |
| } |
| else { |
| oldVal = metaCache.get(new GridCacheInternalKeyImpl(name, grpName)); |
| |
| if (oldVal == null) |
| return null; |
| else if (!(oldVal instanceof DistributedCollectionMetadata)) |
| throw new IgniteCheckedException("Another data structure with the same name already created " + |
| "[name=" + name + |
| ", newType=" + type + |
| ", existingType=" + oldVal.type() + ']'); |
| |
| cache = ctx.cache().getOrStartCache(((DistributedCollectionMetadata)oldVal).cacheName()); |
| |
| if (cache == null) |
| return null; |
| } |
| |
| if (oldVal != null) { |
| if (oldVal.type() != type) |
| throw new IgniteCheckedException("Another data structure with the same name already created " + |
| "[name=" + name + |
| ", newType=" + type + |
| ", existingType=" + oldVal.type() + ']'); |
| |
| assert oldVal instanceof DistributedCollectionMetadata; |
| |
| if (cfg != null && ((DistributedCollectionMetadata)oldVal).configuration().isCollocated() != cfg.isCollocated()) { |
| throw new IgniteCheckedException("Another collection with the same name but different " + |
| "configuration already created [name=" + name + |
| ", newCollocated=" + cfg.isCollocated() + |
| ", existingCollocated=" + !cfg.isCollocated() + ']'); |
| } |
| } |
| |
| return retryTopologySafe(new IgniteOutClosureX<T>() { |
| @Override public T applyx() throws IgniteCheckedException { |
| return c.applyx(cache.context()); |
| } |
| }); |
| } |
| |
| /** |
| * Awaits for processor initialization. |
| */ |
| private void awaitInitialization() { |
| CountDownLatch latch0 = initLatch; |
| |
| if (latch0 == null) |
| throw new IllegalStateException("Ignite cluster is not active"); |
| |
| if (latch0.getCount() > 0) { |
| try { |
| U.await(latch0); |
| |
| if (initFailed) |
| throw new IllegalStateException("Failed to initialize data structures processor."); |
| } |
| catch (IgniteInterruptedCheckedException e) { |
| throw new IllegalStateException("Failed to initialize data structures processor " + |
| "(thread has been interrupted).", e); |
| } |
| } |
| } |
| |
| /** |
| * Gets or creates count down latch. If count down latch is not found in cache, |
| * it is created using provided name and count parameter. |
| * |
| * @param name Name of the latch. |
| * @param cfg Configuration. |
| * @param cnt Initial count. |
| * @param autoDel {@code True} to automatically delete latch from cache when |
| * its count reaches zero. |
| * @param create If {@code true} latch will be created in case it is not in cache, |
| * if it is {@code false} all parameters except {@code name} are ignored. |
| * @return Count down latch for the given name or {@code null} if it is not found and |
| * {@code create} is false. |
| * @throws IgniteCheckedException If operation failed. |
| */ |
| public IgniteCountDownLatch countDownLatch(final String name, |
| @Nullable AtomicConfiguration cfg, |
| final int cnt, |
| final boolean autoDel, |
| final boolean create |
| ) throws IgniteCheckedException { |
| if (create) |
| A.ensure(cnt >= 0, "count can not be negative"); |
| |
| return getAtomic(new AtomicAccessor<GridCacheCountDownLatchEx>() { |
| @Override public T2<GridCacheCountDownLatchEx, AtomicDataStructureValue> get( |
| GridCacheInternalKey key, |
| AtomicDataStructureValue val, |
| IgniteInternalCache cache |
| ) throws IgniteCheckedException { |
| // Check that count down hasn't been created in other thread yet. |
| GridCacheCountDownLatchEx latch = cast(dsMap.get(key), GridCacheCountDownLatchEx.class); |
| |
| if (latch != null) { |
| assert val != null; |
| |
| return new T2<>(latch, null); |
| } |
| |
| if (val == null && !create) |
| return null; |
| |
| GridCacheCountDownLatchValue retVal = (val == null ? new GridCacheCountDownLatchValue(cnt, autoDel, |
| ctx.discovery().gridStartTime()) : null); |
| |
| GridCacheCountDownLatchValue latchVal = retVal != null ? retVal : (GridCacheCountDownLatchValue)val; |
| |
| assert latchVal != null; |
| |
| latch = new GridCacheCountDownLatchImpl(name, latchVal.initialCount(), |
| latchVal.autoDelete(), |
| key, |
| cache); |
| |
| return new T2<GridCacheCountDownLatchEx, AtomicDataStructureValue>(latch, retVal); |
| } |
| }, cfg, name, COUNT_DOWN_LATCH, create, GridCacheCountDownLatchEx.class); |
| } |
| |
| /** |
| * Removes count down latch from cache. |
| * |
| * @param name Name of the latch. |
| * @param grpName Cache group name. |
| * @throws IgniteCheckedException If operation failed. |
| */ |
| public void removeCountDownLatch(final String name, final String grpName) throws IgniteCheckedException { |
| removeDataStructure(new IgnitePredicateX<AtomicDataStructureValue>() { |
| @Override public boolean applyx(AtomicDataStructureValue val) throws IgniteCheckedException { |
| assert val != null && val instanceof GridCacheCountDownLatchValue; |
| |
| GridCacheCountDownLatchValue latchVal = (GridCacheCountDownLatchValue)val; |
| |
| if (latchVal.get() > 0) { |
| throw new IgniteCheckedException("Failed to remove count down latch " + |
| "with non-zero count: " + latchVal.get()); |
| } |
| |
| return true; |
| } |
| }, name, grpName, COUNT_DOWN_LATCH, null); |
| } |
| |
| /** |
| * Gets or creates semaphore. If semaphore is not found in cache, |
| * it is created using provided name and count parameter. |
| * |
| * @param name Name of the semaphore. |
| * @param cfg Configuration. |
| * @param cnt Initial count. |
| * @param failoverSafe {@code True} FailoverSafe parameter. |
| * @param create If {@code true} semaphore will be created in case it is not in cache, |
| * if it is {@code false} all parameters except {@code name} are ignored. |
| * @return Semaphore for the given name or {@code null} if it is not found and |
| * {@code create} is false. |
| * @throws IgniteCheckedException If operation failed. |
| */ |
| public IgniteSemaphore semaphore(final String name, @Nullable AtomicConfiguration cfg, final int cnt, |
| final boolean failoverSafe, final boolean create) |
| throws IgniteCheckedException { |
| return getAtomic(new AtomicAccessor<GridCacheSemaphoreEx>() { |
| @Override public T2<GridCacheSemaphoreEx, AtomicDataStructureValue> get( |
| GridCacheInternalKey key, |
| AtomicDataStructureValue val, |
| IgniteInternalCache cache |
| ) throws IgniteCheckedException { |
| // Check that semaphore hasn't been created in other thread yet. |
| GridCacheSemaphoreEx sem = cast(dsMap.get(key), GridCacheSemaphoreEx.class); |
| |
| if (sem != null) { |
| assert val != null; |
| |
| return new T2<>(sem, null); |
| } |
| |
| if (val == null && !create) |
| return null; |
| |
| AtomicDataStructureValue retVal = (val == null ? new GridCacheSemaphoreState(cnt, |
| new HashMap<UUID, Integer>(), failoverSafe, ctx.discovery().gridStartTime()) : null); |
| |
| GridCacheSemaphoreEx sem0 = new GridCacheSemaphoreImpl(name, key, cache); |
| |
| //check Cluster state against semaphore state |
| if (val != null && failoverSafe) { |
| GridCacheSemaphoreState semState = (GridCacheSemaphoreState)val; |
| |
| boolean updated = false; |
| |
| Map<UUID, Integer> waiters = semState.getWaiters(); |
| |
| Integer permit = ((GridCacheSemaphoreState)val).getCount(); |
| |
| for (UUID nodeId : new HashSet<>(waiters.keySet())) { |
| |
| ClusterNode node = ctx.cluster().get().node(nodeId); |
| |
| if (node == null) { |
| |
| permit += waiters.get(nodeId); |
| |
| waiters.remove(nodeId); |
| |
| updated = true; |
| } |
| } |
| if (updated) { |
| semState.setWaiters(waiters); |
| semState.setCount(permit); |
| |
| retVal = semState; |
| } |
| } |
| |
| return new T2<>(sem0, retVal); |
| } |
| }, cfg, name, SEMAPHORE, create, GridCacheSemaphoreEx.class); |
| } |
| |
| /** |
| * Removes semaphore from cache. |
| * |
| * @param name Name of the semaphore. |
| * @param grpName Group name. |
| * @throws IgniteCheckedException If operation failed. |
| */ |
| public void removeSemaphore(final String name, final String grpName) throws IgniteCheckedException { |
| removeDataStructure(new IgnitePredicateX<AtomicDataStructureValue>() { |
| @Override public boolean applyx(AtomicDataStructureValue val) throws IgniteCheckedException { |
| assert val != null && val instanceof GridCacheSemaphoreState; |
| |
| GridCacheSemaphoreState semVal = (GridCacheSemaphoreState)val; |
| |
| if (semVal.getCount() < 0) |
| throw new IgniteCheckedException("Failed to remove semaphore with blocked threads. "); |
| |
| return true; |
| } |
| }, name, grpName, SEMAPHORE, null); |
| } |
| |
| /** |
| * Gets or creates reentrant lock. If reentrant lock is not found in cache, |
| * it is created using provided name, failover mode, and fairness mode parameters. |
| * |
| * @param name Name of the reentrant lock. |
| * @param cfg Configuration. |
| * @param failoverSafe Flag indicating behaviour in case of failure. |
| * @param fair Flag indicating fairness policy of this lock. |
| * @param create If {@code true} reentrant lock will be created in case it is not in cache. |
| * @return ReentrantLock for the given name or {@code null} if it is not found and |
| * {@code create} is false. |
| * @throws IgniteCheckedException If operation failed. |
| */ |
| public IgniteLock reentrantLock(final String name, @Nullable AtomicConfiguration cfg, final boolean failoverSafe, |
| final boolean fair, final boolean create) throws IgniteCheckedException { |
| return getAtomic(new AtomicAccessor<GridCacheLockEx>() { |
| @Override public T2<GridCacheLockEx, AtomicDataStructureValue> get( |
| GridCacheInternalKey key, |
| AtomicDataStructureValue val, |
| IgniteInternalCache cache |
| ) throws IgniteCheckedException { |
| // Check that reentrant lock hasn't been created in other thread yet. |
| GridCacheLockEx reentrantLock = cast(dsMap.get(key), GridCacheLockEx.class); |
| |
| if (reentrantLock != null) { |
| assert val != null; |
| |
| return new T2<>(reentrantLock, null); |
| } |
| |
| if (val == null && !create) |
| return new T2<>(null, null); |
| |
| AtomicDataStructureValue retVal = (val == null ? new GridCacheLockState(0, ctx.localNodeId(), |
| 0, failoverSafe, fair, ctx.discovery().gridStartTime()) : null); |
| |
| GridCacheLockEx reentrantLock0 = new GridCacheLockImpl(name, key, cache); |
| |
| return new T2<>(reentrantLock0, retVal); |
| } |
| }, cfg, name, REENTRANT_LOCK, create, GridCacheLockEx.class); |
| } |
| |
| /** |
| * Removes reentrant lock from cache. |
| * |
| * @param name Name of the reentrant lock. |
| * @param grpName Group name. |
| * @param broken Flag indicating the reentrant lock is broken and should be removed unconditionally. |
| * @throws IgniteCheckedException If operation failed. |
| */ |
| public void removeReentrantLock(final String name, final String grpName, final boolean broken) throws IgniteCheckedException { |
| removeDataStructure(new IgnitePredicateX<AtomicDataStructureValue>() { |
| @Override public boolean applyx(AtomicDataStructureValue val) throws IgniteCheckedException { |
| assert val != null && val instanceof GridCacheLockState; |
| |
| GridCacheLockState lockVal = (GridCacheLockState)val; |
| |
| if (lockVal.get() > 0 && !broken) |
| throw new IgniteCheckedException("Failed to remove reentrant lock with blocked threads. "); |
| |
| return true; |
| } |
| }, name, grpName, REENTRANT_LOCK, null); |
| } |
| |
| /** |
| * |
| */ |
| static class DataStructuresEntryFilter implements CacheEntryEventSerializableFilter<Object, Object> { |
| /** */ |
| private static final long serialVersionUID = 0L; |
| |
| /** {@inheritDoc} */ |
| @Override public boolean evaluate(CacheEntryEvent<?, ?> evt) throws CacheEntryListenerException { |
| if (evt.getEventType() == EventType.CREATED || evt.getEventType() == EventType.UPDATED) |
| return evt.getValue() instanceof GridCacheCountDownLatchValue || |
| evt.getValue() instanceof GridCacheSemaphoreState || |
| evt.getValue() instanceof GridCacheLockState; |
| else { |
| assert evt.getEventType() == EventType.REMOVED : evt; |
| |
| return true; |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return S.toString(DataStructuresEntryFilter.class, this); |
| } |
| } |
| |
| /** |
| * |
| */ |
| private class DataStructuresEntryListener implements |
| CacheEntryUpdatedListener<GridCacheInternalKey, GridCacheInternal> { |
| /** {@inheritDoc} */ |
| @Override public void onUpdated( |
| Iterable<CacheEntryEvent<? extends GridCacheInternalKey, ? extends GridCacheInternal>> evts |
| ) throws CacheEntryListenerException { |
| for (CacheEntryEvent<? extends GridCacheInternalKey, ? extends GridCacheInternal> evt : evts) { |
| if (evt.getEventType() == EventType.CREATED || evt.getEventType() == EventType.UPDATED) { |
| GridCacheInternal val0 = evt.getValue(); |
| |
| if (val0 instanceof GridCacheCountDownLatchValue) { |
| final GridCacheInternalKey key = evt.getKey(); |
| |
| // Notify latch on changes. |
| final GridCacheRemovable latch = dsMap.get(key); |
| |
| GridCacheCountDownLatchValue val = (GridCacheCountDownLatchValue)val0; |
| |
| if (latch instanceof GridCacheCountDownLatchEx) { |
| final GridCacheCountDownLatchEx latch0 = (GridCacheCountDownLatchEx)latch; |
| |
| latch0.onUpdate(val.get()); |
| |
| if (val.get() == 0 && val.autoDelete()) { |
| dsMap.remove(key); |
| |
| IgniteInternalFuture<?> rmvFut = ctx.closure().runLocalSafe(new GPR() { |
| @Override public void run() { |
| try { |
| removeCountDownLatch(latch0.name(), key.groupName()); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to remove count down latch: " + latch0.name(), e); |
| } |
| finally { |
| ctx.cache().context().txContextReset(); |
| } |
| } |
| }); |
| |
| rmvFut.listen(new CI1<IgniteInternalFuture<?>>() { |
| @Override public void apply(IgniteInternalFuture<?> f) { |
| try { |
| f.get(); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to remove count down latch: " + latch0.name(), e); |
| } |
| |
| latch.onRemoved(); |
| } |
| }); |
| } |
| } |
| else if (latch != null) { |
| U.error(log, "Failed to cast object " + |
| "[expected=" + IgniteCountDownLatch.class.getSimpleName() + |
| ", actual=" + latch.getClass() + ", value=" + latch + ']'); |
| } |
| } |
| else if (val0 instanceof GridCacheSemaphoreState) { |
| GridCacheInternalKey key = evt.getKey(); |
| |
| // Notify semaphore on changes. |
| final GridCacheRemovable sem = dsMap.get(key); |
| |
| GridCacheSemaphoreState val = (GridCacheSemaphoreState)val0; |
| |
| if (sem instanceof GridCacheSemaphoreEx) { |
| final GridCacheSemaphoreEx semaphore0 = (GridCacheSemaphoreEx)sem; |
| |
| semaphore0.onUpdate(val); |
| } |
| else if (sem != null) { |
| U.error(log, "Failed to cast object " + |
| "[expected=" + IgniteSemaphore.class.getSimpleName() + |
| ", actual=" + sem.getClass() + ", value=" + sem + ']'); |
| } |
| } |
| else if (val0 instanceof GridCacheLockState) { |
| GridCacheInternalKey key = evt.getKey(); |
| |
| // Notify reentrant lock on changes. |
| final GridCacheRemovable reentrantLock = dsMap.get(key); |
| |
| GridCacheLockState val = (GridCacheLockState)val0; |
| |
| if (reentrantLock instanceof GridCacheLockEx) { |
| final GridCacheLockEx lock0 = (GridCacheLockEx)reentrantLock; |
| |
| lock0.onUpdate(val); |
| } |
| else if (reentrantLock != null) { |
| U.error(log, "Failed to cast object " + |
| "[expected=" + IgniteLock.class.getSimpleName() + |
| ", actual=" + reentrantLock.getClass() + ", value=" + reentrantLock + ']'); |
| } |
| } |
| } |
| else { |
| assert evt.getEventType() == EventType.REMOVED : evt; |
| |
| GridCacheInternal key = evt.getKey(); |
| |
| // Entry's val is null if entry deleted. |
| GridCacheRemovable obj = dsMap.remove(key); |
| |
| if (obj != null) |
| obj.onRemoved(); |
| } |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return S.toString(DataStructuresEntryListener.class, this); |
| } |
| } |
| |
| /** |
| * Gets a set from cache or creates one if it's not cached. |
| * |
| * @param name Set name. |
| * @param grpName Group name. If present, will override groupName from configuration. |
| * @param cfg Set configuration if new set should be created. |
| * @return Set instance. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @Nullable public <T> IgniteSet<T> set(final String name, @Nullable final String grpName, @Nullable final CollectionConfiguration cfg) |
| throws IgniteCheckedException { |
| A.notNull(name, "name"); |
| |
| final boolean create = cfg != null; |
| final boolean collocated = isCollocated(cfg); |
| final boolean separated = !collocated && |
| U.isOldestNodeVersionAtLeast(SEPARATE_CACHE_PER_NON_COLLOCATED_SET_SINCE, ctx.grid().cluster().nodes()); |
| |
| return getCollection(new CX1<GridCacheContext, IgniteSet<T>>() { |
| @Override public IgniteSet<T> applyx(GridCacheContext cctx) throws IgniteCheckedException { |
| return cctx.dataStructures().set(name, collocated, create, separated); |
| } |
| }, cfg, name, grpName, SET, create, separated); |
| } |
| |
| /** |
| * Gets a set from cache by known cache id. Does not create new sets. |
| * |
| * @param name Set name. |
| * @param cacheId Cache id. |
| * @param collocated Colocated mode flag. |
| * @param separated Separated cache flag. |
| * @return Set instance. |
| * @throws IgniteCheckedException If failed. |
| */ |
| @Nullable public <T> IgniteSet<T> set(String name, int cacheId, boolean collocated, boolean separated) |
| throws IgniteCheckedException { |
| A.notNull(name, "name"); |
| |
| DynamicCacheDescriptor desc = ctx.cache().cacheDescriptor(cacheId); |
| |
| if (desc == null) |
| return null; |
| |
| IgniteInternalCache<Object, Object> cache = ctx.cache().cache(desc.cacheName()); |
| |
| if (cache == null) |
| return null; |
| |
| return cache.context().dataStructures().set(name, collocated, false, separated); |
| } |
| |
| /** |
| * @param name Set name. |
| * @param cctx Set cache context. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public void removeSet(final String name, final GridCacheContext cctx) throws IgniteCheckedException { |
| assert name != null; |
| assert cctx != null; |
| |
| CIX1<GridCacheSetHeader> afterRmv = new CIX1<GridCacheSetHeader>() { |
| @Override public void applyx(GridCacheSetHeader hdr) throws IgniteCheckedException { |
| hdr = (GridCacheSetHeader)cctx.cache().withNoRetries().getAndRemove(new GridCacheSetHeaderKey(name)); |
| |
| if (hdr != null) |
| cctx.dataStructures().removeSetData(hdr.id(), hdr.separated()); |
| } |
| }; |
| |
| removeDataStructure(null, name, cctx.group().name(), SET, afterRmv); |
| } |
| |
| /** |
| * @param log Logger. |
| * @param call Callable. |
| * @return Callable result. |
| * @throws IgniteCheckedException If all retries failed. |
| */ |
| public static <R> R retry(IgniteLogger log, Callable<R> call) throws IgniteCheckedException { |
| try { |
| return GridCacheUtils.retryTopologySafe(call); |
| } |
| catch (IgniteCheckedException e) { |
| throw e; |
| } |
| catch (Exception e) { |
| throw new IgniteCheckedException(e); |
| } |
| } |
| |
| /** |
| * Tries to cast the object to expected type. |
| * |
| * @param obj Object which will be casted. |
| * @param cls Class |
| * @param <R> Type of expected result. |
| * @return Object has casted to expected type. |
| * @throws IgniteCheckedException If {@code obj} has different to {@code cls} type. |
| */ |
| @Nullable private <R> R cast(@Nullable Object obj, Class<R> cls) throws IgniteCheckedException { |
| if (obj == null) |
| return null; |
| |
| if (cls.isInstance(obj)) |
| return (R)obj; |
| else |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void printMemoryStats() { |
| X.println(">>> "); |
| X.println(">>> Data structure processor memory stats [igniteInstanceName=" + ctx.igniteInstanceName() + ']'); |
| X.println(">>> dsMapSize: " + dsMap.size()); |
| } |
| |
| /** |
| * @throws IgniteException If atomics configuration is not provided. |
| */ |
| private void checkAtomicsConfiguration() throws IgniteException { |
| if (dfltAtomicCfg == null) |
| throw new IgniteException("Atomic data structure can not be created, " + |
| "need to provide AtomicConfiguration."); |
| } |
| |
| /** |
| * @param c Closure to run. |
| * @throws IgniteCheckedException If failed. |
| * @return Closure return value. |
| */ |
| private static <T> T retryTopologySafe(IgniteOutClosureX<T> c) throws IgniteCheckedException { |
| for (int i = 0; i < GridCacheAdapter.MAX_RETRIES; i++) { |
| try { |
| return c.applyx(); |
| } |
| catch (NodeStoppingException e) { |
| throw e; |
| } |
| catch (IgniteCheckedException e) { |
| if (i == GridCacheAdapter.MAX_RETRIES - 1) |
| throw e; |
| |
| ClusterTopologyCheckedException topErr = e.getCause(ClusterTopologyCheckedException.class); |
| |
| if (topErr == null || (topErr instanceof ClusterTopologyServerNotFoundException)) |
| throw e; |
| |
| IgniteInternalFuture<?> fut = topErr.retryReadyFuture(); |
| |
| if (fut != null) |
| fut.get(); |
| } |
| } |
| |
| assert false; |
| |
| return null; |
| } |
| |
| /** |
| * |
| */ |
| private interface AtomicAccessor<T> { |
| /** |
| * @param key Key. |
| * @param val Existing value. |
| * @param cache Data structure cache. |
| * @return Data structure instance and value to store in cache. |
| * @throws IgniteCheckedException If failed. |
| */ |
| T2<T, AtomicDataStructureValue> get(GridCacheInternalKey key, |
| @Nullable AtomicDataStructureValue val, |
| IgniteInternalCache cache) throws IgniteCheckedException; |
| } |
| } |
