/*
 * 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.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.KnownVersion;
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;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * 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(GeodeGlossary.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);
      }
    }
  }

  @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 KnownVersion 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(getLastModified());
    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, 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(getKey());
              throw new RegionClearedException(
                  "During the GII put of entry, the region got cleared so aborting the operation");
            }
          }
        }
        setValue(region, 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)
