| /* |
| * 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.geode.internal.cache.entries; |
| |
| import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ABSTRACT_REGION_ENTRY_FILL_IN_VALUE; |
| import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ABSTRACT_REGION_ENTRY_PREPARE_VALUE_FOR_CACHE; |
| |
| import java.io.IOException; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.InvalidDeltaException; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.cache.CacheWriterException; |
| import org.apache.geode.cache.EntryEvent; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.TimeoutException; |
| import org.apache.geode.cache.query.IndexMaintenanceException; |
| import org.apache.geode.cache.query.QueryException; |
| import org.apache.geode.cache.query.internal.index.IndexManager; |
| import org.apache.geode.cache.query.internal.index.IndexProtocol; |
| import org.apache.geode.cache.util.GatewayConflictHelper; |
| import org.apache.geode.cache.util.GatewayConflictResolver; |
| import org.apache.geode.distributed.internal.DistributionConfig; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.HeapDataOutputStream; |
| import org.apache.geode.internal.InternalStatisticsDisabledException; |
| import org.apache.geode.internal.cache.CachedDeserializable; |
| import org.apache.geode.internal.cache.CachedDeserializableFactory; |
| import org.apache.geode.internal.cache.DistributedRegion; |
| import org.apache.geode.internal.cache.EntryEventImpl; |
| import org.apache.geode.internal.cache.FilterProfile; |
| import org.apache.geode.internal.cache.ImageState; |
| import org.apache.geode.internal.cache.InitialImageOperation.Entry; |
| import org.apache.geode.internal.cache.InternalCacheEvent; |
| import org.apache.geode.internal.cache.InternalRegion; |
| import org.apache.geode.internal.cache.RegionClearedException; |
| import org.apache.geode.internal.cache.RegionEntryContext; |
| import org.apache.geode.internal.cache.RegionQueue; |
| import org.apache.geode.internal.cache.TXManagerImpl; |
| import org.apache.geode.internal.cache.TimestampedEntryEventImpl; |
| import org.apache.geode.internal.cache.Token; |
| import org.apache.geode.internal.cache.TombstoneService; |
| import org.apache.geode.internal.cache.ValueComparisonHelper; |
| import org.apache.geode.internal.cache.eviction.EvictionList; |
| import org.apache.geode.internal.cache.persistence.DiskRecoveryStore; |
| import org.apache.geode.internal.cache.persistence.DiskStoreID; |
| import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException; |
| import org.apache.geode.internal.cache.versions.RegionVersionVector; |
| import org.apache.geode.internal.cache.versions.VersionSource; |
| import org.apache.geode.internal.cache.versions.VersionStamp; |
| import org.apache.geode.internal.cache.versions.VersionTag; |
| import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl; |
| import org.apache.geode.internal.lang.StringUtils; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| import org.apache.geode.internal.offheap.MemoryAllocator; |
| import org.apache.geode.internal.offheap.MemoryAllocatorImpl; |
| import org.apache.geode.internal.offheap.OffHeapHelper; |
| import org.apache.geode.internal.offheap.ReferenceCountHelper; |
| import org.apache.geode.internal.offheap.Releasable; |
| import org.apache.geode.internal.offheap.StoredObject; |
| import org.apache.geode.internal.offheap.annotations.Released; |
| import org.apache.geode.internal.offheap.annotations.Retained; |
| import org.apache.geode.internal.offheap.annotations.Unretained; |
| import org.apache.geode.internal.serialization.ByteArrayDataInput; |
| import org.apache.geode.internal.serialization.Version; |
| import org.apache.geode.internal.util.BlobHelper; |
| import org.apache.geode.internal.util.Versionable; |
| import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap; |
| import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntry; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| import org.apache.geode.pdx.PdxInstance; |
| import org.apache.geode.pdx.PdxSerializationException; |
| import org.apache.geode.pdx.internal.ConvertableToBytes; |
| import org.apache.geode.pdx.internal.PdxInstanceImpl; |
| |
| /** |
| * Abstract implementation class of RegionEntry interface. This is the topmost implementation class |
| * so common behavior lives here. |
| * |
| * @since GemFire 3.5.1 |
| */ |
| public abstract class AbstractRegionEntry implements HashRegionEntry<Object, Object> { |
| private static final Logger logger = LogService.getLogger(); |
| |
| /** |
| * Whether to disable last access time update when a put occurs. The default is false (enable last |
| * access time update on put). To disable it, set the 'gemfire.disableAccessTimeUpdateOnPut' |
| * system property. |
| */ |
| protected static final boolean DISABLE_ACCESS_TIME_UPDATE_ON_PUT = |
| Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "disableAccessTimeUpdateOnPut"); |
| |
| /* |
| * Flags for a Region Entry. These flags are stored in the msb of the long used to also store the |
| * lastModificationTime. |
| */ |
| private static final long VALUE_RESULT_OF_SEARCH = 0x01L << 56; |
| |
| private static final long UPDATE_IN_PROGRESS = 0x02L << 56; |
| |
| private static final long LISTENER_INVOCATION_IN_PROGRESS = 0x08L << 56; |
| |
| /** used for LRUEntry instances. */ |
| protected static final long RECENTLY_USED = 0x10L << 56; |
| |
| /** used for LRUEntry instances. */ |
| protected static final long EVICTED = 0x20L << 56; |
| |
| /** |
| * Set if the entry is being used by a transactions. Some features (eviction and expiration) will |
| * not modify an entry when a tx is using it to prevent the tx to fail do to conflict. |
| */ |
| private static final long IN_USE_BY_TX = 0x40L << 56; |
| |
| protected AbstractRegionEntry(RegionEntryContext context, |
| @Retained(ABSTRACT_REGION_ENTRY_PREPARE_VALUE_FOR_CACHE) Object value) { |
| |
| setValue(context, prepareValueForCache(context, value, false), false); |
| |
| // setLastModified(System.currentTimeMillis()); this must be set later so we can use ==0 |
| // to know this is a new entry in checkForConflicts |
| } |
| |
| @Override |
| @SuppressWarnings("IMSE_DONT_CATCH_IMSE") |
| public boolean dispatchListenerEvents(final EntryEventImpl event) throws InterruptedException { |
| final InternalRegion rgn = event.getRegion(); |
| |
| if (event.callbacksInvoked()) { |
| return true; |
| } |
| |
| // don't wait for certain events to reach the head of the queue before |
| // dispatching listeners. However, we must not notify the gateways for |
| // remote-origin ops out of order. Otherwise the other systems will have |
| // inconsistent content. |
| |
| event.setCallbacksInvokedByCurrentThread(); |
| |
| if (logger.isDebugEnabled() && !rgn.isInternalRegion()) { |
| logger.debug("{} dispatching event {}", this, event); |
| } |
| // All the following code that sets "thr" is to workaround |
| // spurious IllegalMonitorStateExceptions caused by JVM bugs. |
| try { |
| // call invokeCallbacks while synced on RegionEntry |
| event.invokeCallbacks(rgn, event.inhibitCacheListenerNotification(), false); |
| return true; |
| |
| } finally { |
| if (isRemoved() && !isTombstone() && !event.isEvicted()) { |
| // Phase 2 of region entry removal is done here. The first phase is done |
| // by the RegionMap. It is unclear why this code is needed. ARM destroy |
| // does this also and we are now doing it as phase3 of the ARM destroy. |
| removePhase2(); |
| ((DiskRecoveryStore) rgn).getRegionMap().removeEntry(event.getKey(), this, true, event, |
| rgn); |
| } |
| } |
| } |
| |
| @Override |
| public long getLastAccessed() throws InternalStatisticsDisabledException { |
| throw new InternalStatisticsDisabledException(); |
| } |
| |
| @Override |
| public long getHitCount() throws InternalStatisticsDisabledException { |
| throw new InternalStatisticsDisabledException(); |
| } |
| |
| @Override |
| public long getMissCount() throws InternalStatisticsDisabledException { |
| throw new InternalStatisticsDisabledException(); |
| } |
| |
| /** |
| * This sets the lastModified time for the entry. In subclasses with statistics it will also set |
| * the lastAccessed time unless the system property gemfire.disableAccessTimeUpdateOnPut is set to |
| * true. |
| * |
| * @param lastModified the time of last modification of the entry |
| */ |
| public void setLastModified(long lastModified) { |
| setLastModifiedAndAccessedTimes(lastModified, lastModified); |
| } |
| |
| /** |
| * This sets the lastModified and lastAccessed time for the entry. Subclasses that do not keep |
| * track of lastAccessed time will ignore the second parameter. |
| * |
| * @param lastModified the time of last modification of the entry |
| * @param lastAccessed the time the entry was last accessed |
| */ |
| protected void setLastModifiedAndAccessedTimes(long lastModified, long lastAccessed) { |
| _setLastModified(lastModified); |
| } |
| |
| @Override |
| public void txDidDestroy(long currentTime) { |
| setLastModifiedAndAccessedTimes(currentTime, currentTime); |
| } |
| |
| @Override |
| public void updateStatsForPut(long lastModifiedTime, long lastAccessedTime) { |
| setLastModifiedAndAccessedTimes(lastModifiedTime, lastAccessedTime); |
| } |
| |
| @Override |
| public void setRecentlyUsed(RegionEntryContext context) { |
| // do nothing by default; only needed for LRU |
| } |
| |
| @Override |
| public void updateStatsForGet(boolean hit, long time) { |
| // nothing needed |
| } |
| |
| @Override |
| public void resetCounts() throws InternalStatisticsDisabledException { |
| throw new InternalStatisticsDisabledException(); |
| } |
| |
| void _removePhase1() { |
| _setValue(Token.REMOVED_PHASE1); |
| } |
| |
| @Override |
| public void removePhase1(InternalRegion region, boolean clear) throws RegionClearedException { |
| _removePhase1(); |
| } |
| |
| @Override |
| public void removePhase2() { |
| _setValue(Token.REMOVED_PHASE2); |
| } |
| |
| @Override |
| public void makeTombstone(InternalRegion region, VersionTag version) |
| throws RegionClearedException { |
| assert region.getVersionVector() != null; |
| assert version != null; |
| |
| boolean wasTombstone = isTombstone(); |
| setRecentlyUsed(region); |
| boolean newEntry = getValueAsToken() == Token.REMOVED_PHASE1; |
| basicMakeTombstone(region); |
| region.scheduleTombstone(this, version, wasTombstone); |
| if (newEntry) { |
| // bug #46631 - entry count is decremented by scheduleTombstone but this is a new entry |
| region.getCachePerfStats().incEntryCount(1); |
| } |
| } |
| |
| private void basicMakeTombstone(InternalRegion region) throws RegionClearedException { |
| boolean setValueCompleted = false; |
| try { |
| setValue(region, Token.TOMBSTONE); |
| setValueCompleted = true; |
| } finally { |
| if (!setValueCompleted && isTombstone()) { |
| removePhase2(); |
| } |
| } |
| } |
| |
| @Override |
| public void setValueWithTombstoneCheck(@Unretained Object v, EntryEvent e) |
| throws RegionClearedException { |
| if (v == Token.TOMBSTONE) { |
| makeTombstone((InternalRegion) e.getRegion(), ((InternalCacheEvent) e).getVersionTag()); |
| } else { |
| setValue((RegionEntryContext) e.getRegion(), v, (EntryEventImpl) e); |
| } |
| } |
| |
| /** |
| * Return true if the object is removed. |
| * |
| * TODO this method does NOT return true if the object is Token.DESTROYED. dispatchListenerEvents |
| * relies on that fact to avoid removing destroyed tokens from the map. We should refactor so that |
| * this method calls Token.isRemoved, and places that don't want a destroyed Token can explicitly |
| * check for a DESTROY token. |
| */ |
| @Override |
| public boolean isRemoved() { |
| Token o = getValueAsToken(); |
| return o == Token.REMOVED_PHASE1 || o == Token.REMOVED_PHASE2 || o == Token.TOMBSTONE; |
| } |
| |
| @Override |
| public boolean isDestroyedOrRemoved() { |
| return Token.isRemoved(getValueAsToken()); |
| } |
| |
| @Override |
| public boolean isDestroyedOrRemovedButNotTombstone() { |
| Token o = getValueAsToken(); |
| return o == Token.DESTROYED || o == Token.REMOVED_PHASE1 || o == Token.REMOVED_PHASE2; |
| } |
| |
| @Override |
| public boolean isTombstone() { |
| return getValueAsToken() == Token.TOMBSTONE; |
| } |
| |
| @Override |
| public boolean isRemovedPhase2() { |
| return getValueAsToken() == Token.REMOVED_PHASE2; |
| } |
| |
| @Override |
| public boolean fillInValue(InternalRegion region, |
| @Retained(ABSTRACT_REGION_ENTRY_FILL_IN_VALUE) Entry entry, ByteArrayDataInput in, |
| DistributionManager mgr, final Version version) { |
| |
| // starting default value |
| entry.setSerialized(false); |
| |
| @Retained(ABSTRACT_REGION_ENTRY_FILL_IN_VALUE) |
| final Object v; |
| if (isTombstone()) { |
| v = Token.TOMBSTONE; |
| } else { |
| // OFFHEAP: need to incrc, copy bytes, decrc |
| v = getValue(region); |
| if (v == null) { |
| return false; |
| } |
| } |
| |
| entry.setLastModified(mgr, getLastModified()); // fix for bug 31059 |
| if (v == Token.INVALID) { |
| entry.setInvalid(); |
| } else if (v == Token.LOCAL_INVALID) { |
| entry.setLocalInvalid(); |
| } else if (v == Token.TOMBSTONE) { |
| entry.setTombstone(); |
| } else if (v instanceof CachedDeserializable) { |
| // don't serialize here if it is not already serialized |
| CachedDeserializable cd = (CachedDeserializable) v; |
| if (!cd.isSerialized()) { |
| entry.setValue(cd.getDeserializedForReading()); |
| } else { |
| Object tmp = cd.getValue(); |
| if (tmp instanceof byte[]) { |
| entry.setValue(tmp); |
| } else { |
| try { |
| HeapDataOutputStream hdos = new HeapDataOutputStream(version); |
| BlobHelper.serializeTo(tmp, hdos); |
| hdos.trim(); |
| entry.setValue(hdos); |
| } catch (IOException e) { |
| throw new IllegalArgumentException( |
| "An IOException was thrown while serializing.", |
| e); |
| } |
| } |
| entry.setSerialized(true); |
| } |
| } else if (v instanceof byte[]) { |
| entry.setValue(v); |
| } else { |
| Object preparedValue = v; |
| if (preparedValue != null) { |
| preparedValue = prepareValueForGII(preparedValue); |
| if (preparedValue == null) { |
| return false; |
| } |
| } |
| try { |
| HeapDataOutputStream hdos = new HeapDataOutputStream(version); |
| BlobHelper.serializeTo(preparedValue, hdos); |
| hdos.trim(); |
| entry.setValue(hdos); |
| entry.setSerialized(true); |
| } catch (IOException e) { |
| throw new IllegalArgumentException( |
| "An IOException was thrown while serializing.", |
| e); |
| } |
| } |
| return true; |
| } |
| |
| /** |
| * To fix bug 49901 if v is a GatewaySenderEventImpl then make a heap copy of it if it is offheap. |
| * |
| * @return the value to provide to the gii request; null if no value should be provided. |
| */ |
| static Object prepareValueForGII(Object v) { |
| assert v != null; |
| if (v instanceof GatewaySenderEventImpl) { |
| return ((GatewaySenderEventImpl) v).makeHeapCopyIfOffHeap(); |
| } else { |
| return v; |
| } |
| } |
| |
| @Override |
| public boolean isOverflowedToDisk(InternalRegion region, |
| DistributedRegion.DiskPosition diskPosition) { |
| return false; |
| } |
| |
| @Override |
| public Object getValue(RegionEntryContext context) { |
| ReferenceCountHelper.createReferenceCountOwner(); |
| @Retained |
| Object result = getValueRetain(context, true); |
| |
| // If the thread is an Index Creation Thread & the value obtained is |
| // Token.REMOVED , we can skip synchronization block. This is required to prevent |
| // the dead lock caused if an Index Update Thread has gone into a wait holding the |
| // lock of the Entry object. There should not be an issue if the Index creation thread |
| // gets the temporary value of token.REMOVED as the correct value will get indexed |
| // by the Index Update Thread , once the index creation thread has exited. |
| // Part of Bugfix # 33336 |
| |
| if (Token.isRemoved(result)) { |
| ReferenceCountHelper.setReferenceCountOwner(null); |
| return null; |
| } else { |
| result = OffHeapHelper.copyAndReleaseIfNeeded(result, context.getCache()); |
| ReferenceCountHelper.setReferenceCountOwner(null); |
| setRecentlyUsed(context); |
| return result; |
| } |
| } |
| |
| @Override |
| @Retained |
| public Object getValueRetain(RegionEntryContext context) { |
| @Retained |
| Object result = getValueRetain(context, true); |
| if (Token.isRemoved(result)) { |
| return null; |
| } else { |
| setRecentlyUsed(context); |
| return result; |
| } |
| } |
| |
| @Override |
| @Released |
| public void setValue(RegionEntryContext context, @Unretained Object value) |
| throws RegionClearedException { |
| // TODO: This will mark new entries as being recently used |
| // It might be better to only mark them when they are modified. |
| // Or should we only mark them on reads? |
| setValue(context, value, true); |
| } |
| |
| @Override |
| public void setValue(RegionEntryContext context, Object value, EntryEventImpl event) |
| throws RegionClearedException { |
| setValue(context, value); |
| } |
| |
| @Released |
| protected void setValue(RegionEntryContext context, @Unretained Object value, |
| boolean recentlyUsed) { |
| _setValue(value); |
| releaseOffHeapRefIfRegionBeingClosedOrDestroyed(context, value); |
| if (recentlyUsed) { |
| setRecentlyUsed(context); |
| } |
| } |
| |
| void releaseOffHeapRefIfRegionBeingClosedOrDestroyed(RegionEntryContext context, Object ref) { |
| if (isOffHeapReference(ref) && isThisRegionBeingClosedOrDestroyed(context)) { |
| ((Releasable) this).release(); |
| } |
| } |
| |
| private boolean isThisRegionBeingClosedOrDestroyed(RegionEntryContext context) { |
| return context instanceof InternalRegion |
| && ((InternalRegion) context).isThisRegionBeingClosedOrDestroyed(); |
| } |
| |
| private boolean isOffHeapReference(Object ref) { |
| return ref != Token.REMOVED_PHASE1 && this instanceof OffHeapRegionEntry |
| && ref instanceof StoredObject && ((StoredObject) ref).hasRefCount(); |
| } |
| |
| /** |
| * This method determines if the value is in a compressed representation and decompresses it if it |
| * is. |
| * |
| * @param context the values context. |
| * @param value a region entry value. |
| * |
| * @return the decompressed form of the value parameter. |
| */ |
| static Object decompress(RegionEntryContext context, Object value) { |
| if (isCompressible(context, value)) { |
| long time = context.getCachePerfStats().startDecompression(); |
| value = EntryEventImpl.deserialize(context.getCompressor().decompress((byte[]) value)); |
| context.getCachePerfStats().endDecompression(time); |
| } |
| |
| return value; |
| } |
| |
| protected static Object compress(RegionEntryContext context, Object value) { |
| return compress(context, value, null); |
| } |
| |
| /** |
| * This method determines if the value is compressible and compresses it if it is. |
| * |
| * @param context the values context. |
| * @param value a region entry value. |
| * |
| * @return the compressed form of the value parameter. |
| */ |
| protected static Object compress(RegionEntryContext context, Object value, EntryEventImpl event) { |
| if (isCompressible(context, value)) { |
| long time = context.getCachePerfStats().startCompression(); |
| byte[] serializedValue; |
| if (event != null && event.getCachedSerializedNewValue() != null) { |
| serializedValue = event.getCachedSerializedNewValue(); |
| if (value instanceof CachedDeserializable) { |
| CachedDeserializable cd = (CachedDeserializable) value; |
| if (!(cd.getValue() instanceof byte[])) { |
| // The cd now has the object form so use the cached serialized form in a new cd. |
| // This serialization is much cheaper than reserializing the object form. |
| serializedValue = EntryEventImpl |
| .serialize(CachedDeserializableFactory.create(serializedValue, context.getCache())); |
| } else { |
| serializedValue = EntryEventImpl.serialize(cd); |
| } |
| } |
| } else { |
| serializedValue = EntryEventImpl.serialize(value); |
| if (event != null && !(value instanceof byte[])) { |
| // See if we can cache the serialized new value in the event. |
| // If value is a byte[] then we don't cache it since it is not serialized. |
| if (value instanceof CachedDeserializable) { |
| // For a CacheDeserializable we want to only cache the wrapped value; |
| // not the serialized CacheDeserializable. |
| CachedDeserializable cd = (CachedDeserializable) value; |
| Object cdVal = cd.getValue(); |
| if (cdVal instanceof byte[]) { |
| event.setCachedSerializedNewValue((byte[]) cdVal); |
| } |
| } else { |
| event.setCachedSerializedNewValue(serializedValue); |
| } |
| } |
| } |
| value = context.getCompressor().compress(serializedValue); |
| context.getCachePerfStats().endCompression(time, serializedValue.length, |
| ((byte[]) value).length); |
| } |
| |
| return value; |
| } |
| |
| private static byte[] compressBytes(RegionEntryContext context, byte[] uncompressedBytes) { |
| byte[] result = uncompressedBytes; |
| if (isCompressible(context, uncompressedBytes)) { |
| long time = context.getCachePerfStats().startCompression(); |
| result = context.getCompressor().compress(uncompressedBytes); |
| context.getCachePerfStats().endCompression(time, uncompressedBytes.length, result.length); |
| } |
| return result; |
| } |
| |
| |
| @Override |
| public Object getValueInVM(RegionEntryContext context) { |
| ReferenceCountHelper.createReferenceCountOwner(); |
| @Released |
| Object v = getValueRetain(context, true); |
| |
| if (v == null) { |
| // should only be possible if disk entry |
| v = Token.NOT_AVAILABLE; |
| } |
| Object result = OffHeapHelper.copyAndReleaseIfNeeded(v, context.getCache()); |
| ReferenceCountHelper.setReferenceCountOwner(null); |
| return result; |
| } |
| |
| @Override |
| public Object getValueInVMOrDiskWithoutFaultIn(InternalRegion region) { |
| return getValueInVM(region); |
| } |
| |
| @Override |
| @Retained |
| public Object getValueOffHeapOrDiskWithoutFaultIn(InternalRegion region) { |
| @Retained |
| Object result = getValueRetain(region, true); |
| return result; |
| } |
| |
| @Override |
| public Object getValueOnDisk(InternalRegion region) throws EntryNotFoundException { |
| throw new IllegalStateException( |
| "Cannot get value on disk for a region that does not access the disk."); |
| } |
| |
| @Override |
| public Object getSerializedValueOnDisk(final InternalRegion region) |
| throws EntryNotFoundException { |
| throw new IllegalStateException( |
| "Cannot get value on disk for a region that does not access the disk."); |
| } |
| |
| @Override |
| public Object getValueOnDiskOrBuffer(InternalRegion region) throws EntryNotFoundException { |
| throw new IllegalStateException( |
| "Cannot get value on disk for a region that does not access the disk."); |
| // TODO: if value is Token.REMOVED || Token.DESTROYED throw EntryNotFoundException |
| } |
| |
| @Override |
| public boolean initialImagePut(final InternalRegion region, final long lastModified, |
| Object newValue, boolean wasRecovered, boolean acceptedVersionTag) |
| throws RegionClearedException { |
| // note that the caller has already write synced this RegionEntry |
| return initialImageInit(region, lastModified, newValue, this.isTombstone(), wasRecovered, |
| acceptedVersionTag); |
| } |
| |
| @Override |
| public boolean initialImageInit(final InternalRegion region, final long lastModified, |
| final Object newValue, final boolean create, final boolean wasRecovered, |
| final boolean acceptedVersionTag) throws RegionClearedException { |
| |
| // note that the caller has already write synced this RegionEntry |
| boolean result = false; |
| |
| // if it has been destroyed then don't do anything |
| Token vTok = getValueAsToken(); |
| if (acceptedVersionTag || create || (vTok != Token.DESTROYED || vTok != Token.TOMBSTONE)) { |
| // OFFHEAP noop |
| Object newValueToWrite = newValue; |
| // OFFHEAP noop |
| boolean putValue = acceptedVersionTag || create || (newValueToWrite != Token.LOCAL_INVALID |
| && (wasRecovered || (vTok == Token.LOCAL_INVALID))); |
| |
| if (region.isUsedForPartitionedRegionAdmin() |
| && newValueToWrite instanceof CachedDeserializable) { |
| // Special case for partitioned region meta data |
| // We do not need the RegionEntry on this case. |
| // Because the pr meta data region will not have an LRU. |
| newValueToWrite = |
| ((CachedDeserializable) newValueToWrite).getDeserializedValue(region, null); |
| if (!create && newValueToWrite instanceof Versionable) { |
| // Heap value should always be deserialized at this point // OFFHEAP will not be |
| // deserialized |
| final Object oldValue = getValueInVM(region); |
| // BUGFIX for 35029. If oldValue is null the newValue should be put. |
| if (oldValue == null) { |
| putValue = true; |
| } else if (oldValue instanceof Versionable) { |
| Versionable nv = (Versionable) newValueToWrite; |
| Versionable ov = (Versionable) oldValue; |
| putValue = nv.isNewerThan(ov); |
| } |
| } |
| } |
| |
| if (putValue) { |
| // change to INVALID if region itself has been invalidated, |
| // and current value is recovered |
| if (create || acceptedVersionTag) { |
| // At this point, since we now always recover from disk first, |
| // we only care about "isCreate" since "isRecovered" is impossible |
| // if we had a regionInvalidate or regionClear |
| ImageState imageState = region.getImageState(); |
| // this method is called during loadSnapshot as well as getInitialImage |
| if (imageState.getRegionInvalidated()) { |
| if (newValueToWrite != Token.TOMBSTONE) { |
| newValueToWrite = Token.INVALID; |
| } |
| } else if (imageState.getClearRegionFlag()) { |
| boolean entryOK = false; |
| RegionVersionVector rvv = imageState.getClearRegionVersionVector(); |
| if (rvv != null) { // a filtered clear |
| VersionSource id = getVersionStamp().getMemberID(); |
| if (id == null) { |
| id = region.getVersionMember(); |
| } |
| if (!rvv.contains(id, getVersionStamp().getRegionVersion())) { |
| entryOK = true; |
| } |
| } |
| if (!entryOK) { |
| // If the region has been issued cleared during |
| // the GII , then those entries loaded before this one would have |
| // been cleared from the Map due to clear operation & for the |
| // currententry whose key may have escaped the clearance , will be |
| // cleansed by the destroy token. |
| newValueToWrite = Token.DESTROYED; // TODO: never used |
| imageState.addDestroyedEntry(this.getKey()); |
| throw new RegionClearedException( |
| "During the GII put of entry, the region got cleared so aborting the operation"); |
| } |
| } |
| } |
| setValue(region, this.prepareValueForCache(region, newValueToWrite, false)); |
| result = true; |
| |
| if (newValueToWrite != Token.TOMBSTONE) { |
| if (create) { |
| region.getCachePerfStats().incCreates(); |
| } |
| region.updateStatsForPut(this, lastModified, false); |
| } |
| |
| if (logger.isTraceEnabled()) { |
| if (newValueToWrite instanceof CachedDeserializable) { |
| logger.trace("ProcessChunk: region={}; put a CachedDeserializable ({},{})", |
| region.getFullPath(), getKey(), |
| ((CachedDeserializable) newValueToWrite).getStringForm()); |
| } else { |
| logger.trace("ProcessChunk: region={}; put({},{})", region.getFullPath(), getKey(), |
| StringUtils.forceToString(newValueToWrite)); |
| } |
| } |
| } |
| } |
| return result; |
| } |
| |
| /** |
| * @throws EntryNotFoundException if expectedOldValue is not null and is not equal to current |
| * value |
| */ |
| @Override |
| @Released |
| public boolean destroy(InternalRegion region, EntryEventImpl event, boolean inTokenMode, |
| boolean cacheWrite, @Unretained Object expectedOldValue, boolean forceDestroy, |
| boolean removeRecoveredEntry) throws CacheWriterException, EntryNotFoundException, |
| TimeoutException, RegionClearedException { |
| |
| // A design decision was made to not retrieve the old value from the disk |
| // if the entry has been evicted to only have the CacheListener afterDestroy |
| // method ignore it. We don't want to pay the performance penalty. The |
| // getValueInVM method does not retrieve the value from disk if it has been |
| // evicted. Instead, it uses the NotAvailable token. |
| // |
| // If the region is a WAN queue region, the old value is actually used by the |
| // afterDestroy callback on a secondary. It is not needed on a primary. |
| // Since the destroy that sets WAN_QUEUE_TOKEN always originates on the primary |
| // we only pay attention to WAN_QUEUE_TOKEN if the event is originRemote. |
| // |
| // We also read old value from disk or buffer |
| // in the case where there is a non-null expectedOldValue |
| // see PartitionedRegion#remove(Object key, Object value) |
| ReferenceCountHelper.skipRefCountTracking(); |
| @Retained |
| @Released |
| Object curValue = getValueRetain(region, true); |
| ReferenceCountHelper.unskipRefCountTracking(); |
| boolean proceed; |
| try { |
| if (curValue == null) { |
| curValue = Token.NOT_AVAILABLE; |
| } |
| |
| if (curValue == Token.NOT_AVAILABLE) { |
| // In some cases we need to get the current value off of disk. |
| |
| // if the event is transmitted during GII and has an old value, it was |
| // the state of the transmitting cache's entry & should be used here |
| if (event.getCallbackArgument() != null |
| && event.getCallbackArgument().equals(RegionQueue.WAN_QUEUE_TOKEN) |
| && event.isOriginRemote()) { // check originRemote for bug 40508 |
| // curValue = getValue(region); can cause deadlock if GII is occurring |
| curValue = getValueOnDiskOrBuffer(region); |
| } else { |
| FilterProfile fp = region.getFilterProfile(); |
| if (fp != null && (fp.getCqCount() > 0 || expectedOldValue != null)) { |
| // curValue = getValue(region); can cause deadlock will fault in the value |
| // and will confuse LRU. |
| curValue = getValueOnDiskOrBuffer(region); |
| } |
| } |
| } |
| |
| if (expectedOldValue != null) { |
| if (!checkExpectedOldValue(expectedOldValue, curValue, region)) { |
| throw new EntryNotFoundException( |
| "The current value was not equal to expected value."); |
| } |
| } |
| |
| if (inTokenMode && event.hasOldValue()) { |
| proceed = true; |
| } else { |
| event.setOldValue(curValue, curValue instanceof GatewaySenderEventImpl); |
| proceed = region.getConcurrencyChecksEnabled() || removeRecoveredEntry || forceDestroy |
| || destroyShouldProceedBasedOnCurrentValue(curValue) |
| || (event.getOperation() == Operation.REMOVE && (curValue == null |
| || curValue == Token.LOCAL_INVALID || curValue == Token.INVALID)); |
| } |
| } finally { |
| OffHeapHelper.releaseWithNoTracking(curValue); |
| } |
| |
| if (proceed) { |
| // Generate the version tag if needed. This method should only be |
| // called if we are in fact going to destroy the entry, so it must be |
| // after the entry not found exception above. |
| if (!removeRecoveredEntry) { |
| region.generateAndSetVersionTag(event, this); |
| } |
| if (cacheWrite) { |
| region.cacheWriteBeforeDestroy(event, expectedOldValue); |
| if (event.getRegion().getServerProxy() != null) { |
| // server will return a version tag |
| // update version information (may throw ConcurrentCacheModificationException) |
| VersionStamp stamp = getVersionStamp(); |
| if (stamp != null) { |
| stamp.processVersionTag(event); |
| } |
| } |
| } |
| region.recordEvent(event); |
| // don't do index maintenance on a destroy if the value in the |
| // RegionEntry (the old value) is invalid |
| updateIndexOnDestroyOperation(region); |
| |
| boolean removeEntry = false; |
| VersionTag v = event.getVersionTag(); |
| if (region.getConcurrencyChecksEnabled() && !removeRecoveredEntry |
| && !event.isFromRILocalDestroy()) { |
| // bug #46780, don't retain tombstones for entries destroyed for register-interest |
| // Destroy will write a tombstone instead |
| if (v == null || !v.hasValidVersion()) { |
| // localDestroy and eviction and ops received with no version tag |
| // should create a tombstone using the existing version stamp, as should |
| // (bug #45245) responses from servers that do not have valid version information |
| VersionStamp stamp = this.getVersionStamp(); |
| if (stamp != null) { // proxy has no stamps |
| v = stamp.asVersionTag(); |
| event.setVersionTag(v); |
| } |
| } |
| removeEntry = v == null || !v.hasValidVersion(); |
| } else { |
| removeEntry = true; |
| } |
| |
| if (removeEntry) { |
| boolean isThisTombstone = isTombstone(); |
| if (inTokenMode && !event.getOperation().isEviction()) { |
| setValue(region, Token.DESTROYED); |
| } else { |
| removePhase1(region, false); |
| } |
| if (isThisTombstone) { |
| region.unscheduleTombstone(this); |
| } |
| } else { |
| makeTombstone(region, v); |
| } |
| |
| return true; |
| } else { |
| return false; |
| } |
| } |
| |
| protected void updateIndexOnDestroyOperation(InternalRegion region) { |
| if (!isTombstone() && !region.isProxy() && !isInvalid()) { |
| IndexManager indexManager = region.getIndexManager(); |
| if (indexManager != null) { |
| try { |
| if (isValueNull()) { |
| @Released |
| Object value = getValueOffHeapOrDiskWithoutFaultIn(region); |
| try { |
| Object preparedValue = prepareValueForCache(region, value, false); |
| _setValue(preparedValue); |
| releaseOffHeapRefIfRegionBeingClosedOrDestroyed(region, preparedValue); |
| } finally { |
| OffHeapHelper.release(value); |
| } |
| } |
| indexManager.updateIndexes(this, IndexManager.REMOVE_ENTRY, IndexProtocol.OTHER_OP); |
| } catch (QueryException e) { |
| throw new IndexMaintenanceException(e); |
| } |
| } |
| } |
| } |
| |
| private static boolean destroyShouldProceedBasedOnCurrentValue(Object curValue) { |
| if (curValue == null) { |
| return false; |
| } |
| if (Token.isRemoved(curValue)) { |
| return false; |
| } |
| return true; |
| } |
| |
| public static boolean checkExpectedOldValue(@Unretained Object expectedOldValue, |
| @Unretained Object actualValue, InternalRegion region) { |
| |
| if (Token.isInvalid(expectedOldValue)) { |
| return actualValue == null || Token.isInvalid(actualValue); |
| } else { |
| boolean isCompressedOffHeap = |
| region.getAttributes().getOffHeap() && region.getAttributes().getCompressor() != null; |
| return ValueComparisonHelper |
| .checkEquals(expectedOldValue, actualValue, isCompressedOffHeap, region.getCache()); |
| } |
| } |
| |
| // Do not add any instance fields to this class. |
| // Instead add them to LeafRegionEntry.cpp |
| |
| public static class HashRegionEntryCreator |
| implements CustomEntryConcurrentHashMap.HashEntryCreator<Object, Object> { |
| |
| @Override |
| public HashEntry<Object, Object> newEntry(final Object key, final int hash, |
| final HashEntry<Object, Object> next, final Object value) { |
| final AbstractRegionEntry entry = (AbstractRegionEntry) value; |
| // if hash is already set then assert that the two should be same |
| final int entryHash = entry.getEntryHash(); |
| if (hash == 0 || entryHash != 0) { |
| if (entryHash != hash) { |
| Assert.fail("unexpected mismatch of hash, expected=" + hash + ", actual=" + entryHash |
| + " for " + entry); |
| } |
| } |
| entry.setEntryHash(hash); |
| entry.setNextEntry(next); |
| return entry; |
| } |
| |
| @Override |
| public int keyHashCode(final Object key, final boolean compareValues) { |
| return CustomEntryConcurrentHashMap.keyHash(key, compareValues); |
| } |
| } |
| |
| @Override |
| public abstract Object getKey(); |
| |
| private static boolean okToStoreOffHeap(Object v, AbstractRegionEntry e) { |
| if (v == null) |
| return false; |
| if (Token.isInvalidOrRemoved(v)) |
| return false; |
| if (v == Token.NOT_AVAILABLE) |
| return false; |
| if (v instanceof DiskEntry.RecoveredEntry) |
| return false; // The disk layer has special logic that ends up storing the nested value in the |
| // RecoveredEntry off heap |
| if (!(e instanceof OffHeapRegionEntry)) |
| return false; |
| // TODO should we check for deltas here or is that a user error? |
| return true; |
| } |
| |
| /** |
| * Default implementation. Override in subclasses with primitive keys to prevent creating an |
| * Object form of the key for each equality check. |
| */ |
| @Override |
| public boolean isKeyEqual(Object k) { |
| return k.equals(getKey()); |
| } |
| |
| private static final long LAST_MODIFIED_MASK = 0x00FFFFFFFFFFFFFFL; |
| |
| protected void _setLastModified(long lastModifiedTime) { |
| if (lastModifiedTime < 0 || lastModifiedTime > LAST_MODIFIED_MASK) { |
| throw new IllegalStateException("Expected lastModifiedTime " + lastModifiedTime |
| + " to be >= 0 and <= " + LAST_MODIFIED_MASK); |
| } |
| long storedValue; |
| long newValue; |
| do { |
| storedValue = getLastModifiedField(); |
| newValue = storedValue & ~LAST_MODIFIED_MASK; |
| newValue |= lastModifiedTime; |
| } while (!compareAndSetLastModifiedField(storedValue, newValue)); |
| } |
| |
| protected abstract long getLastModifiedField(); |
| |
| protected abstract boolean compareAndSetLastModifiedField(long expectedValue, long newValue); |
| |
| @Override |
| public long getLastModified() { |
| return getLastModifiedField() & LAST_MODIFIED_MASK; |
| } |
| |
| protected boolean areAnyBitsSet(long bitMask) { |
| return (getLastModifiedField() & bitMask) != 0L; |
| } |
| |
| /** |
| * Any bits in "bitMask" that are 1 will be set. |
| */ |
| protected void setBits(long bitMask) { |
| boolean done; |
| do { |
| long bits = getLastModifiedField(); |
| long newBits = bits | bitMask; |
| if (bits == newBits) |
| return; |
| done = compareAndSetLastModifiedField(bits, newBits); |
| } while (!done); |
| } |
| |
| /** |
| * Any bits in "bitMask" that are 0 will be cleared. |
| */ |
| protected void clearBits(long bitMask) { |
| boolean done; |
| do { |
| long bits = getLastModifiedField(); |
| long newBits = bits & bitMask; |
| if (bits == newBits) |
| return; |
| done = compareAndSetLastModifiedField(bits, newBits); |
| } while (!done); |
| } |
| |
| @Override |
| @Retained(ABSTRACT_REGION_ENTRY_PREPARE_VALUE_FOR_CACHE) |
| public Object prepareValueForCache(RegionEntryContext r, |
| @Retained(ABSTRACT_REGION_ENTRY_PREPARE_VALUE_FOR_CACHE) Object val, boolean isEntryUpdate) { |
| return prepareValueForCache(r, val, null, isEntryUpdate); |
| } |
| |
| @Override |
| @Retained(ABSTRACT_REGION_ENTRY_PREPARE_VALUE_FOR_CACHE) |
| public Object prepareValueForCache(RegionEntryContext r, |
| @Retained(ABSTRACT_REGION_ENTRY_PREPARE_VALUE_FOR_CACHE) Object val, EntryEventImpl event, |
| boolean isEntryUpdate) { |
| if (r != null && r.getOffHeap() && okToStoreOffHeap(val, this)) { |
| if (val instanceof StoredObject) { |
| // Check to see if val has the same compression settings as this region. |
| // The recursive calls in this section are safe because |
| // we only do it after copy the off-heap value to the heap. |
| // This is needed to fix bug 52057. |
| StoredObject soVal = (StoredObject) val; |
| assert !soVal.isCompressed(); |
| if (r.getCompressor() != null) { |
| // val is uncompressed and we need a compressed value. |
| // So copy the off-heap value to the heap in a form that can be compressed. |
| byte[] valAsBytes = soVal.getValueAsHeapByteArray(); |
| Object heapValue; |
| if (soVal.isSerialized()) { |
| heapValue = CachedDeserializableFactory.create(valAsBytes, r.getCache()); |
| } else { |
| heapValue = valAsBytes; |
| } |
| return prepareValueForCache(r, heapValue, event, isEntryUpdate); |
| } |
| if (soVal.hasRefCount()) { |
| // if the reused StoredObject has a refcount then need to increment it |
| if (!soVal.retain()) { |
| throw new IllegalStateException("Could not use an off heap value because it was freed"); |
| } |
| } |
| // else it is has no refCount so just return it as prepared. |
| } else { |
| byte[] data; |
| boolean isSerialized = !(val instanceof byte[]); |
| if (isSerialized) { |
| if (event != null && event.getCachedSerializedNewValue() != null) { |
| data = event.getCachedSerializedNewValue(); |
| } else { |
| if (val instanceof CachedDeserializable) { |
| data = ((CachedDeserializable) val).getSerializedValue(); |
| } else if (val instanceof PdxInstance) { |
| try { |
| data = ((ConvertableToBytes) val).toBytes(); |
| } catch (IOException e) { |
| throw new PdxSerializationException("Could not convert " + val + " to bytes", e); |
| } |
| } else { |
| data = EntryEventImpl.serialize(val); |
| } |
| if (event != null) { |
| event.setCachedSerializedNewValue(data); |
| } |
| } |
| } else { |
| data = (byte[]) val; |
| } |
| byte[] compressedData = compressBytes(r, data); |
| // TODO: array comparison is broken |
| boolean isCompressed = compressedData != data; |
| ReferenceCountHelper.setReferenceCountOwner(this); |
| MemoryAllocator ma = MemoryAllocatorImpl.getAllocator(); // fix for bug 47875 |
| val = ma.allocateAndInitialize(compressedData, isSerialized, isCompressed, data); |
| ReferenceCountHelper.setReferenceCountOwner(null); |
| } |
| return val; |
| } |
| @Unretained |
| Object nv = val; |
| if (nv instanceof StoredObject) { |
| // This off heap value is being put into a on heap region. |
| byte[] data = ((StoredObject) nv).getSerializedValue(); |
| nv = CachedDeserializableFactory.create(data, r.getCache()); |
| } |
| if (nv instanceof PdxInstanceImpl) { |
| // We do not want to put PDXs in the cache as values. |
| // So get the serialized bytes and use a CachedDeserializable. |
| try { |
| byte[] data = ((ConvertableToBytes) nv).toBytes(); |
| byte[] compressedData = compressBytes(r, data); |
| // TODO: array comparison is broken |
| if (data == compressedData) { |
| nv = CachedDeserializableFactory.create(data, r.getCache()); |
| } else { |
| nv = compressedData; |
| } |
| } catch (IOException e) { |
| throw new PdxSerializationException("Could not convert " + nv + " to bytes", e); |
| } |
| } else { |
| nv = compress(r, nv, event); |
| } |
| return nv; |
| } |
| |
| @Override |
| @Unretained |
| public Object getValue() { |
| return getValueField(); |
| } |
| |
| @Override |
| public boolean isUpdateInProgress() { |
| return areAnyBitsSet(UPDATE_IN_PROGRESS); |
| } |
| |
| @Override |
| public void setUpdateInProgress(final boolean underUpdate) { |
| if (underUpdate) { |
| setBits(UPDATE_IN_PROGRESS); |
| } else { |
| clearBits(~UPDATE_IN_PROGRESS); |
| } |
| } |
| |
| @Override |
| public boolean isCacheListenerInvocationInProgress() { |
| return areAnyBitsSet(LISTENER_INVOCATION_IN_PROGRESS); |
| } |
| |
| @Override |
| public void setCacheListenerInvocationInProgress(final boolean isListenerInvoked) { |
| if (isListenerInvoked) { |
| setBits(LISTENER_INVOCATION_IN_PROGRESS); |
| } else { |
| clearBits(~LISTENER_INVOCATION_IN_PROGRESS); |
| } |
| } |
| |
| @Override |
| public synchronized boolean isInUseByTransaction() { |
| return areAnyBitsSet(IN_USE_BY_TX); |
| } |
| |
| private void setInUseByTransaction(final boolean v) { |
| if (v) { |
| setBits(IN_USE_BY_TX); |
| } else { |
| clearBits(~IN_USE_BY_TX); |
| } |
| } |
| |
| @Override |
| public synchronized void incRefCount() { |
| TXManagerImpl.incRefCount(this); |
| setInUseByTransaction(true); |
| } |
| |
| @Override |
| public synchronized void decRefCount(EvictionList evictionList, InternalRegion region) { |
| if (TXManagerImpl.decRefCount(this)) { |
| if (isInUseByTransaction()) { |
| setInUseByTransaction(false); |
| if (!isDestroyedOrRemoved()) { |
| appendToEvictionList(evictionList); |
| if (region != null && region.isEntryExpiryPossible()) { |
| region.addExpiryTaskIfAbsent(this); |
| } |
| } |
| } |
| } |
| } |
| |
| @Override |
| public synchronized void resetRefCount(EvictionList evictionList) { |
| if (isInUseByTransaction()) { |
| setInUseByTransaction(false); |
| appendToEvictionList(evictionList); |
| } |
| } |
| |
| protected void appendToEvictionList(EvictionList evictionList) { |
| // nothing |
| } |
| |
| void _setValue(Object val) { |
| setValueField(val); |
| } |
| |
| @Override |
| public Token getValueAsToken() { |
| Object v = getValueField(); |
| if (v == null || v instanceof Token) { |
| return (Token) v; |
| } else { |
| return Token.NOT_A_TOKEN; |
| } |
| } |
| |
| /** |
| * Reads the value of this region entry. Provides low level access to the value field. |
| * |
| * @return possible OFF_HEAP_OBJECT (caller uses region entry reference) |
| */ |
| @Unretained |
| protected abstract Object getValueField(); |
| |
| /** |
| * Set the value of this region entry. Provides low level access to the value field. |
| * |
| * @param v the new value to set |
| */ |
| protected abstract void setValueField(@Unretained Object v); |
| |
| @Override |
| @Retained |
| public Object getTransformedValue() { |
| return getValueRetain(null, false); |
| } |
| |
| @Override |
| public boolean getValueWasResultOfSearch() { |
| return areAnyBitsSet(VALUE_RESULT_OF_SEARCH); |
| } |
| |
| @Override |
| public void setValueResultOfSearch(boolean v) { |
| if (v) { |
| setBits(VALUE_RESULT_OF_SEARCH); |
| } else { |
| clearBits(~VALUE_RESULT_OF_SEARCH); |
| } |
| } |
| |
| public boolean hasValidVersion() { |
| VersionStamp stamp = (VersionStamp) this; |
| return stamp.getRegionVersion() != 0 || stamp.getEntryVersion() != 0; |
| } |
| |
| @Override |
| public boolean hasStats() { |
| // override this in implementations that have stats |
| return false; |
| } |
| |
| @Override |
| public Object getMapValue() { |
| return this; |
| } |
| |
| @Override |
| public void setMapValue(final Object newValue) { |
| if (this != newValue) { |
| Assert.fail("AbstractRegionEntry#setMapValue: unexpected setMapValue " + "with newValue=" |
| + newValue + ", this=" + this); |
| } |
| } |
| |
| protected abstract void setEntryHash(int v); |
| |
| @Override |
| public String toString() { |
| final StringBuilder sb = new StringBuilder(this.getClass().getSimpleName()).append('@') |
| .append(Integer.toHexString(System.identityHashCode(this))).append(" ("); |
| return appendFieldsToString(sb).append(')').toString(); |
| } |
| |
| protected StringBuilder appendFieldsToString(final StringBuilder sb) { |
| // OFFHEAP _getValue ok: the current toString on ObjectChunk is safe to use without incing |
| // refcount. |
| sb.append("key=").append(getKey()).append("; rawValue=").append(getValue()); |
| VersionStamp stamp = getVersionStamp(); |
| if (stamp != null) { |
| sb.append("; version=").append(stamp.asVersionTag()).append(";member=") |
| .append(stamp.getMemberID()); |
| } |
| return sb; |
| } |
| |
| /** |
| * This generates version tags for outgoing messages for all subclasses supporting concurrency |
| * versioning. It also sets the entry's version stamp to the tag's values. |
| */ |
| @Override |
| public VersionTag generateVersionTag(VersionSource member, boolean withDelta, |
| InternalRegion region, EntryEventImpl event) { |
| VersionStamp stamp = this.getVersionStamp(); |
| if (stamp != null && region.getServerProxy() == null) { |
| // clients do not generate versions |
| int v = stamp.getEntryVersion() + 1; |
| if (v > 0xFFFFFF) { |
| v -= 0x1000000; // roll-over |
| } |
| VersionSource previous = stamp.getMemberID(); |
| |
| // For non persistent regions, we allow the member to be null and |
| // when we send a message and the remote side can determine the member |
| // from the sender. For persistent regions, we need to send |
| // the persistent id to the remote side. |
| // |
| // TODO - RVV - optimize the way we send the persistent id to save |
| // space. |
| if (member == null) { |
| VersionSource regionMember = region.getVersionMember(); |
| if (regionMember instanceof DiskStoreID) { |
| member = regionMember; |
| } |
| } |
| |
| VersionTag tag = VersionTag.create(member); |
| tag.setEntryVersion(v); |
| if (region.getVersionVector() != null) { |
| // Use region version if already provided, else generate |
| long nextRegionVersion = event.getNextRegionVersion(); |
| if (nextRegionVersion != -1) { |
| // Set on the tag and record it locally |
| tag.setRegionVersion(nextRegionVersion); |
| RegionVersionVector rvv = region.getVersionVector(); |
| rvv.recordVersion(rvv.getOwnerId(), nextRegionVersion); |
| if (logger.isDebugEnabled()) { |
| logger.debug("recorded region version {}; region={}", nextRegionVersion, |
| region.getFullPath()); |
| } |
| } else { |
| tag.setRegionVersion(region.getVersionVector().getNextVersion()); |
| } |
| } |
| if (withDelta) { |
| tag.setPreviousMemberID(previous); |
| } |
| VersionTag remoteTag = event.getVersionTag(); |
| if (remoteTag != null && remoteTag.isGatewayTag()) { |
| // if this event was received from a gateway we use the remote system's |
| // timestamp and dsid. |
| tag.setVersionTimeStamp(remoteTag.getVersionTimeStamp()); |
| tag.setDistributedSystemId(remoteTag.getDistributedSystemId()); |
| tag.setAllowedByResolver(remoteTag.isAllowedByResolver()); |
| } else { |
| long time = region.cacheTimeMillis(); |
| int dsid = region.getDistributionManager().getDistributedSystemId(); |
| // a locally generated change should always have a later timestamp than |
| // one received from a wan gateway, so fake a timestamp if necessary |
| if (time <= stamp.getVersionTimeStamp() && dsid != tag.getDistributedSystemId()) { |
| time = stamp.getVersionTimeStamp() + 1; |
| } |
| tag.setVersionTimeStamp(time); |
| tag.setDistributedSystemId(dsid); |
| } |
| stamp.setVersions(tag); |
| stamp.setMemberID(member); |
| event.setVersionTag(tag); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "generated tag {}; key={}; oldvalue={} newvalue={} client={} region={}; rvv={}", tag, |
| event.getKey(), event.getOldValueStringForm(), event.getNewValueStringForm(), |
| event.getContext() == null ? "none" |
| : event.getContext().getDistributedMember().getName(), |
| region.getFullPath(), region.getVersionVector()); |
| } |
| return tag; |
| } |
| return null; |
| } |
| |
| /** |
| * This performs a concurrency check. |
| * |
| * This check compares the version number first, followed by the member ID. |
| * |
| * Wraparound of the version number is detected and handled by extending the range of versions by |
| * one bit. |
| * |
| * The normal membership ID comparison method is used. |
| * <p> |
| * |
| * Note that a tag from a remote (WAN) system may be in the event. If this is the case this method |
| * will either invoke a user plugin that allows/disallows the event (and may modify the value) or |
| * it determines whether to allow or disallow the event based on timestamps and |
| * distributedSystemIDs. |
| * |
| * @throws ConcurrentCacheModificationException if the event conflicts with an event that has |
| * already been applied to the entry. |
| */ |
| public void processVersionTag(EntryEvent cacheEvent) { |
| processVersionTag(cacheEvent, true); |
| } |
| |
| protected void processVersionTag(EntryEvent cacheEvent, boolean conflictCheck) { |
| EntryEventImpl event = (EntryEventImpl) cacheEvent; |
| VersionTag tag = event.getVersionTag(); |
| if (tag == null) { |
| return; |
| } |
| |
| try { |
| if (tag.isGatewayTag()) { |
| // this may throw ConcurrentCacheModificationException or modify the event |
| if (processGatewayTag(cacheEvent)) { |
| return; |
| } |
| assert false : "processGatewayTag failure - returned false"; |
| } |
| |
| if (!tag.isFromOtherMember()) { |
| if (!event.getOperation().isNetSearch()) { |
| // except for netsearch, all locally-generated tags can be ignored |
| return; |
| } |
| } |
| |
| final InternalDistributedMember originator = |
| (InternalDistributedMember) event.getDistributedMember(); |
| final VersionSource dmId = event.getRegion().getVersionMember(); |
| InternalRegion r = event.getRegion(); |
| boolean eventHasDelta = event.getDeltaBytes() != null && event.getRawNewValue() == null; |
| |
| VersionStamp stamp = getVersionStamp(); |
| // bug #46223, an event received from a peer or a server may be from a different |
| // distributed system than the last modification made to this entry so we must |
| // perform a gateway conflict check |
| if (stamp != null && !tag.isAllowedByResolver()) { |
| int stampDsId = stamp.getDistributedSystemId(); |
| int tagDsId = tag.getDistributedSystemId(); |
| |
| if (stampDsId != 0 && stampDsId != tagDsId && stampDsId != -1) { |
| StringBuilder verbose = null; |
| if (logger.isTraceEnabled(LogMarker.TOMBSTONE_VERBOSE)) { |
| verbose = new StringBuilder(); |
| verbose.append("processing tag for key ").append(getKey()).append(", stamp=") |
| .append(stamp.asVersionTag()).append(", tag=").append(tag); |
| } |
| long stampTime = stamp.getVersionTimeStamp(); |
| long tagTime = tag.getVersionTimeStamp(); |
| if (stampTime > 0 && (tagTime > stampTime || (tagTime == stampTime |
| && tag.getDistributedSystemId() >= stamp.getDistributedSystemId()))) { |
| if (verbose != null) { |
| verbose.append(" - allowing event"); |
| logger.trace(LogMarker.TOMBSTONE_VERBOSE, verbose); |
| } |
| // Update the stamp with event's version information. |
| applyVersionTag(r, stamp, tag, originator); |
| return; |
| } |
| |
| if (stampTime > 0) { |
| if (verbose != null) { |
| verbose.append(" - disallowing event"); |
| logger.trace(LogMarker.TOMBSTONE_VERBOSE, verbose); |
| } |
| r.getCachePerfStats().incConflatedEventsCount(); |
| persistConflictingTag(r, tag); |
| throw new ConcurrentCacheModificationException("conflicting event detected"); |
| } |
| } |
| } |
| |
| if (r.getVersionVector() != null && r.getServerProxy() == null |
| && (r.getDataPolicy().withPersistence() || !r.getScope().isLocal())) { |
| // bug #45258 - perf degradation for local regions and RVV |
| VersionSource who = tag.getMemberID(); |
| if (who == null) { |
| who = originator; |
| } |
| r.getVersionVector().recordVersion(who, tag); |
| } |
| |
| assert !tag.isFromOtherMember() |
| || tag.getMemberID() != null : "remote tag is missing memberID"; |
| |
| // for a long time I had conflict checks turned off in clients when |
| // receiving a response from a server and applying it to the cache. This lowered |
| // the CPU cost of versioning but eventually had to be pulled for bug #45453 |
| |
| // events coming from servers while a local sync is held on the entry |
| // do not require a conflict check. Conflict checks were already |
| // performed on the server and here we just consume whatever was sent back. |
| // Event.isFromServer() returns true for client-update messages and |
| // for putAll/getAll, which do not hold syncs during the server operation. |
| |
| // for a very long time we had conflict checks turned off for PR buckets. |
| // Bug 45669 showed a primary dying in the middle of distribution. This caused |
| // one backup bucket to have a v2. The other bucket was promoted to primary and |
| // generated a conflicting v2. We need to do the check so that if this second |
| // v2 loses to the original one in the delta-GII operation that the original v2 |
| // will be the winner in both buckets. |
| |
| // The new value in event is not from GII, even it could be tombstone |
| basicProcessVersionTag(r, tag, false, eventHasDelta, dmId, originator, conflictCheck); |
| } catch (ConcurrentCacheModificationException ex) { |
| event.isConcurrencyConflict(true); |
| throw ex; |
| } |
| } |
| |
| protected void basicProcessVersionTag(InternalRegion region, VersionTag tag, |
| boolean isTombstoneFromGII, boolean deltaCheck, VersionSource dmId, |
| InternalDistributedMember sender, boolean checkForConflict) { |
| |
| if (tag != null) { |
| VersionStamp stamp = getVersionStamp(); |
| |
| StringBuilder verbose = null; |
| if (logger.isTraceEnabled(LogMarker.TOMBSTONE_VERBOSE)) { |
| VersionTag stampTag = stamp.asVersionTag(); |
| if (stampTag.hasValidVersion() && checkForConflict) { |
| // only be verbose here if there's a possibility we might reject the operation |
| verbose = new StringBuilder(); |
| verbose.append("processing tag for key ").append(getKey()).append(", stamp=") |
| .append(stamp.asVersionTag()).append(", tag=").append(tag) |
| .append(", checkForConflict=").append(checkForConflict); |
| } |
| } |
| |
| if (stamp == null) { |
| throw new IllegalStateException( |
| "message contained a version tag but this region has no version storage"); |
| } |
| |
| boolean apply = true; |
| |
| try { |
| if (checkForConflict) { |
| apply = checkForConflict(region, stamp, tag, isTombstoneFromGII, deltaCheck, dmId, sender, |
| verbose); |
| } |
| } catch (ConcurrentCacheModificationException e) { |
| // Even if we don't apply the operation we should always retain the |
| // highest timestamp in order for WAN conflict checks to work correctly |
| // because the operation may have been sent to other systems and been |
| // applied there |
| if (!tag.isGatewayTag() && stamp.getDistributedSystemId() == tag.getDistributedSystemId() |
| && tag.getVersionTimeStamp() > stamp.getVersionTimeStamp()) { |
| stamp.setVersionTimeStamp(tag.getVersionTimeStamp()); |
| tag.setTimeStampApplied(true); |
| if (verbose != null) { |
| verbose |
| .append("\nThough in conflict the tag timestamp was more recent and was recorded."); |
| } |
| } |
| throw e; |
| } finally { |
| if (verbose != null) { |
| logger.trace(LogMarker.TOMBSTONE_VERBOSE, verbose); |
| } |
| } |
| |
| if (apply) { |
| applyVersionTag(region, stamp, tag, sender); |
| } |
| } |
| } |
| |
| private void applyVersionTag(InternalRegion region, VersionStamp stamp, VersionTag tag, |
| InternalDistributedMember sender) { |
| VersionSource mbr = tag.getMemberID(); |
| if (mbr == null) { |
| mbr = sender; |
| } |
| mbr = region.getVersionVector().getCanonicalId(mbr); |
| tag.setMemberID(mbr); |
| stamp.setVersions(tag); |
| if (tag.hasPreviousMemberID()) { |
| if (tag.getPreviousMemberID() == null) { |
| tag.setPreviousMemberID(stamp.getMemberID()); |
| } else { |
| tag.setPreviousMemberID( |
| region.getVersionVector().getCanonicalId(tag.getPreviousMemberID())); |
| } |
| } |
| } |
| |
| /** perform conflict checking for a stamp/tag */ |
| private boolean checkForConflict(InternalRegion region, VersionStamp stamp, VersionTag tag, |
| boolean isTombstoneFromGII, boolean deltaCheck, VersionSource dmId, |
| InternalDistributedMember sender, StringBuilder verbose) { |
| |
| int stampVersion = stamp.getEntryVersion(); |
| int tagVersion = tag.getEntryVersion(); |
| |
| if (stamp.getVersionTimeStamp() != 0) { // new entries have no timestamp |
| // check for wrap-around on the version number |
| long difference = tagVersion - stampVersion; |
| if (0x10000 < difference || difference < -0x10000) { |
| if (verbose != null) { |
| verbose.append("\nversion rollover detected: tag=").append(tagVersion).append(" stamp=") |
| .append(stampVersion); |
| } |
| if (difference < 0) { |
| tagVersion += 0x1000000; |
| } else { |
| stampVersion += 0x1000000; |
| } |
| } |
| } |
| if (verbose != null) { |
| verbose.append("\nstamp=v").append(stampVersion).append(" tag=v").append(tagVersion); |
| } |
| |
| if (deltaCheck) { |
| checkForDeltaConflict(region, stampVersion, tagVersion, stamp, tag, dmId, sender, verbose); |
| } |
| |
| boolean throwex = false; |
| boolean apply = false; |
| if (stampVersion == 0 || stampVersion < tagVersion) { |
| if (verbose != null) { |
| verbose.append(" - applying change"); |
| } |
| apply = true; |
| } else if (stampVersion > tagVersion) { |
| if (overwritingOldTombstone(region, stamp, tag, verbose) |
| && tag.getVersionTimeStamp() > stamp.getVersionTimeStamp()) { |
| apply = true; |
| } else { |
| // check for an incoming expired tombstone from an initial image chunk. |
| if (tagVersion > 0 |
| && isExpiredTombstone(region, tag.getVersionTimeStamp(), isTombstoneFromGII) |
| && tag.getVersionTimeStamp() > stamp.getVersionTimeStamp()) { |
| // A special case to apply: when remote entry is expired tombstone, then let local vs |
| // remote with newer timestamp to win |
| if (verbose != null) { |
| verbose.append(" - applying change in Delta GII"); |
| } |
| apply = true; |
| } else { |
| if (verbose != null) { |
| verbose.append(" - disallowing"); |
| } |
| throwex = true; |
| } |
| } |
| } else { |
| if (overwritingOldTombstone(region, stamp, tag, verbose)) { |
| apply = true; |
| } else { |
| // compare member IDs |
| VersionSource stampID = stamp.getMemberID(); |
| if (stampID == null) { |
| stampID = dmId; |
| } |
| VersionSource tagID = tag.getMemberID(); |
| if (tagID == null) { |
| tagID = sender; |
| } |
| if (verbose != null) { |
| verbose.append("\ncomparing IDs"); |
| } |
| int compare = stampID.compareTo(tagID); |
| if (compare < 0) { |
| if (verbose != null) { |
| verbose.append(" - applying change"); |
| } |
| apply = true; |
| } else if (compare > 0) { |
| if (verbose != null) { |
| verbose.append(" - disallowing"); |
| } |
| throwex = true; |
| } else if (tag.isPosDup()) { |
| if (verbose != null) { |
| verbose.append(" - disallowing duplicate marked with posdup"); |
| } |
| throwex = true; |
| } else { |
| if (verbose != null) { |
| verbose.append(" - allowing duplicate"); |
| } |
| } |
| } |
| } |
| |
| if (!apply && throwex) { |
| region.getCachePerfStats().incConflatedEventsCount(); |
| persistConflictingTag(region, tag); |
| throw new ConcurrentCacheModificationException(); |
| } |
| |
| return apply; |
| } |
| |
| private boolean isExpiredTombstone(InternalRegion region, long timestamp, boolean isTombstone) { |
| return isTombstone |
| && timestamp + TombstoneService.REPLICATE_TOMBSTONE_TIMEOUT <= region.cacheTimeMillis(); |
| } |
| |
| private boolean overwritingOldTombstone(InternalRegion region, VersionStamp stamp, VersionTag tag, |
| StringBuilder verbose) { |
| // Tombstone GC does not use locking to stop operations when old tombstones |
| // are being removed. Because of this we might get an operation that was applied |
| // in another VM that has just reaped a tombstone and is now using a reset |
| // entry version number. Because of this we check the timestamp on the current |
| // local entry and see if it is old enough to have expired. If this is the case |
| // we accept the change and allow the tag to be recorded |
| long stampTime = stamp.getVersionTimeStamp(); |
| if (isExpiredTombstone(region, stampTime, this.isTombstone())) { |
| // no local change since the tombstone would have timed out - accept the change |
| if (verbose != null) { |
| verbose.append(" - accepting because local timestamp is old"); |
| } |
| return true; |
| } else { |
| return false; |
| } |
| } |
| |
| protected void persistConflictingTag(InternalRegion region, VersionTag tag) { |
| // only persist region needs to persist conflict tag |
| } |
| |
| /** |
| * for an event containing a delta we must check to see if the tag's previous member id is the |
| * stamp's member id and ensure that the version is only incremented by 1. Otherwise the delta is |
| * being applied to a value that does not match the source of the delta. |
| */ |
| private void checkForDeltaConflict(InternalRegion region, long stampVersion, long tagVersion, |
| VersionStamp stamp, VersionTag tag, VersionSource dmId, InternalDistributedMember sender, |
| StringBuilder verbose) { |
| |
| if (tagVersion != stampVersion + 1) { |
| if (verbose != null) { |
| verbose.append("\ndelta requires full value due to version mismatch"); |
| } |
| region.getCachePerfStats().incDeltaFailedUpdates(); |
| throw new InvalidDeltaException("delta cannot be applied due to version mismatch"); |
| |
| } else { |
| // make sure the tag was based on the value in this entry by checking the |
| // tag's previous-changer ID against this stamp's current ID |
| VersionSource stampID = stamp.getMemberID(); |
| if (stampID == null) { |
| stampID = dmId; |
| } |
| VersionSource tagID = tag.getPreviousMemberID(); |
| if (tagID == null) { |
| tagID = sender; |
| } |
| if (!tagID.equals(stampID)) { |
| if (verbose != null) { |
| verbose.append("\ndelta requires full value. tag.previous=").append(tagID) |
| .append(" but stamp.current=").append(stampID); |
| } |
| region.getCachePerfStats().incDeltaFailedUpdates(); |
| throw new InvalidDeltaException("delta cannot be applied due to version ID mismatch"); |
| } |
| } |
| } |
| |
| private boolean processGatewayTag(EntryEvent cacheEvent) { |
| // Gateway tags are installed in the server-side InternalRegion cache |
| // modification methods. They do not have version numbers or distributed |
| // member IDs. Instead they only have timestamps and distributed system IDs. |
| |
| // If there is a resolver plug-in, invoke it. Otherwise we use the timestamps and |
| // distributed system IDs to determine whether to allow the event to proceed. |
| |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| if (this.isRemoved() && !this.isTombstone()) { |
| return true; // no conflict on a new entry |
| } |
| EntryEventImpl event = (EntryEventImpl) cacheEvent; |
| VersionTag tag = event.getVersionTag(); |
| long stampTime = getVersionStamp().getVersionTimeStamp(); |
| long tagTime = tag.getVersionTimeStamp(); |
| int stampDsid = getVersionStamp().getDistributedSystemId(); |
| int tagDsid = tag.getDistributedSystemId(); |
| if (isDebugEnabled) { |
| logger.debug( |
| "processing gateway version information for {}. Stamp dsid={} time={} Tag dsid={} time={}", |
| event.getKey(), stampDsid, stampTime, tagDsid, tagTime); |
| } |
| if (tagTime == VersionTag.ILLEGAL_VERSION_TIMESTAMP) { |
| return true; // no timestamp received from other system - just apply it |
| } |
| // According to GatewayConflictResolver's java doc, it will only be used on tag with different |
| // distributed system id than stamp's |
| if (stampDsid == -1) { |
| return true; |
| } else if (tagDsid == stampDsid) { |
| if (tagTime >= stampTime) { |
| return true; |
| } else { |
| throw new ConcurrentCacheModificationException("conflicting WAN event detected"); |
| } |
| } |
| GatewayConflictResolver resolver = event.getRegion().getCache().getGatewayConflictResolver(); |
| if (resolver != null) { |
| if (isDebugEnabled) { |
| logger.debug("invoking gateway conflict resolver"); |
| } |
| final boolean[] disallow = new boolean[1]; |
| final Object[] newValue = new Object[] {this}; |
| GatewayConflictHelper helper = new GatewayConflictHelper() { |
| @Override |
| public void disallowEvent() { |
| disallow[0] = true; |
| } |
| |
| @Override |
| public void changeEventValue(Object value) { |
| newValue[0] = value; |
| } |
| }; |
| |
| @Released |
| TimestampedEntryEventImpl timestampedEvent = (TimestampedEntryEventImpl) event |
| .getTimestampedEvent(tagDsid, stampDsid, tagTime, stampTime); |
| |
| // gateway conflict resolvers will usually want to see the old value |
| if (!timestampedEvent.hasOldValue() && isRemoved()) { |
| // OFFHEAP: since isRemoved I think getValue will never be stored off heap in this case |
| timestampedEvent.setOldValue(getValue(timestampedEvent.getRegion())); |
| } |
| |
| Throwable thr = null; |
| try { |
| resolver.onEvent(timestampedEvent, helper); |
| } catch (CancelException cancelled) { |
| throw cancelled; |
| } catch (VirtualMachineError err) { |
| SystemFailure.initiateFailure(err); |
| // If this ever returns, rethrow the error. We're poisoned |
| // now, so don't let this thread continue. |
| throw err; |
| } catch (Throwable t) { |
| // Whenever you catch Error or Throwable, you must also |
| // catch VirtualMachineError (see above). However, there is |
| // _still_ a possibility that you are dealing with a cascading |
| // error condition, so you also need to check to see if the JVM |
| // is still usable: |
| SystemFailure.checkFailure(); |
| logger.error("Exception occurred in GatewayConflictResolver", t); |
| thr = t; |
| } finally { |
| timestampedEvent.release(); |
| } |
| |
| if (isDebugEnabled) { |
| logger.debug("done invoking resolver", thr); |
| } |
| if (thr == null) { |
| if (disallow[0]) { |
| if (isDebugEnabled) { |
| logger.debug("conflict resolver rejected the event for {}", event.getKey()); |
| } |
| throw new ConcurrentCacheModificationException( |
| "WAN conflict resolver rejected the operation"); |
| } |
| |
| tag.setAllowedByResolver(true); |
| |
| if (newValue[0] != this) { |
| if (isDebugEnabled) { |
| logger.debug("conflict resolver changed the value of the event for {}", event.getKey()); |
| } |
| // the resolver changed the event value! |
| event.setNewValue(newValue[0]); |
| } |
| // if nothing was done then we allow the event |
| if (isDebugEnabled) { |
| logger.debug("change was allowed by conflict resolver: {}", tag); |
| } |
| return true; |
| } |
| } |
| if (isDebugEnabled) { |
| logger.debug("performing normal WAN conflict check"); |
| } |
| if (tagTime > stampTime || tagTime == stampTime && tagDsid >= stampDsid) { |
| if (isDebugEnabled) { |
| logger.debug("allowing event"); |
| } |
| return true; |
| } |
| if (isDebugEnabled) { |
| logger.debug("disallowing event for {}", event.getKey()); |
| } |
| throw new ConcurrentCacheModificationException("conflicting WAN event detected"); |
| } |
| |
| static boolean isCompressible(RegionEntryContext context, Object value) { |
| return value != null && context != null && context.getCompressor() != null |
| && !Token.isInvalidOrRemoved(value); |
| } |
| |
| /* subclasses supporting versions must override this */ |
| @Override |
| public VersionStamp getVersionStamp() { |
| return null; |
| } |
| |
| @Override |
| public boolean isValueNull() { |
| return null == getValueAsToken(); |
| } |
| |
| @Override |
| public boolean isInvalid() { |
| return Token.isInvalid(getValueAsToken()); |
| } |
| |
| @Override |
| public boolean isDestroyed() { |
| return Token.isDestroyed(getValueAsToken()); |
| } |
| |
| @Override |
| public void setValueToNull() { |
| _setValue(null); |
| } |
| |
| @Override |
| public boolean isInvalidOrRemoved() { |
| return Token.isInvalidOrRemoved(getValueAsToken()); |
| } |
| |
| /** |
| * This is only retained in off-heap subclasses. However, it's marked as Retained here so that |
| * callers are aware that the value may be retained. |
| */ |
| @Override |
| @Retained |
| public Object getValueRetain(RegionEntryContext context, boolean decompress) { |
| if (decompress) { |
| return decompress(context, getValue()); |
| } else { |
| return getValue(); |
| } |
| } |
| |
| @Override |
| public void returnToPool() { |
| // noop by default |
| } |
| |
| @Override |
| public boolean isEvicted() { |
| return false; |
| } |
| } |
