/*
 * 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 java.io.IOException;
import java.nio.ByteBuffer;

import org.apache.logging.log4j.Logger;

import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.ByteArrayDataInput;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.Version;
import org.apache.geode.internal.cache.AbstractDiskRegion;
import org.apache.geode.internal.cache.BucketRegion;
import org.apache.geode.internal.cache.CachedDeserializable;
import org.apache.geode.internal.cache.CachedDeserializableFactory;
import org.apache.geode.internal.cache.DiskId;
import org.apache.geode.internal.cache.DiskRegion;
import org.apache.geode.internal.cache.DiskStoreImpl.AsyncDiskEntry;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.EntryBits;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.InitialImageOperation;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.PlaceHolderDiskRegion;
import org.apache.geode.internal.cache.RegionClearedException;
import org.apache.geode.internal.cache.RegionEntry;
import org.apache.geode.internal.cache.RegionEntryContext;
import org.apache.geode.internal.cache.RegionMap;
import org.apache.geode.internal.cache.Token;
import org.apache.geode.internal.cache.eviction.EvictableEntry;
import org.apache.geode.internal.cache.eviction.EvictionController;
import org.apache.geode.internal.cache.persistence.BytesAndBits;
import org.apache.geode.internal.cache.persistence.DiskRecoveryStore;
import org.apache.geode.internal.cache.persistence.DiskRegionView;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.offheap.AddressableMemoryManager;
import org.apache.geode.internal.offheap.OffHeapHelper;
import org.apache.geode.internal.offheap.ReferenceCountHelper;
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.util.BlobHelper;

/**
 * Represents an entry in an {@link RegionMap} whose value may be stored on disk. This interface
 * provides accessor and mutator methods for a disk entry's state. This allows us to abstract all of
 * the interesting behavior into a {@linkplain DiskEntry.Helper helper class} that we only need to
 * implement once.
 * <p>
 * Each {@code DiskEntry} has a unique {@code id} that is used by the {@link DiskRegion} to identify
 * the key/value pair. Before the disk entry is written to disk, the value of the {@code id} is
 * {@link DiskRegion#INVALID_ID invalid}. Once the object has been written to disk, the {@code id}
 * is a positive number. If the value is {@linkplain Helper#update updated}, then the {@code id} is
 * negated to signify that the value on disk is dirty.
 *
 * @see DiskRegion
 * @since GemFire 3.2
 */
public interface DiskEntry extends RegionEntry {
  /**
   * Sets the value with a {@link RegionEntryContext}.
   *
   * @param context the value's context.
   * @param value an entry value.
   */
  void setValueWithContext(RegionEntryContext context, @Unretained Object value);

  /**
   * In some cases we need to do something just before we drop the value from a DiskEntry that is
   * being moved (i.e. overflowed) to disk.
   */
  void handleValueOverflow(RegionEntryContext context);

  /**
   * Returns true if the DiskEntry value is equal to {@link Token#DESTROYED},
   * {@link Token#REMOVED_PHASE1}, or {@link Token#REMOVED_PHASE2}.
   */
  boolean isRemovedFromDisk();

  /**
   * Returns the id of this {@code DiskEntry}
   */
  DiskId getDiskId();

  int updateAsyncEntrySize(EvictionController capacityController);

  DiskEntry getPrev();

  DiskEntry getNext();

  void setPrev(DiskEntry v);

  void setNext(DiskEntry v);

  /**
   * Used as the entry value if it was invalidated.
   */
  @Immutable
  byte[] INVALID_BYTES = new byte[0];
  /**
   * Used as the entry value if it was locally invalidated.
   */
  @Immutable
  byte[] LOCAL_INVALID_BYTES = new byte[0];
  /**
   * Used as the entry value if it was tombstone.
   */
  @Immutable
  byte[] TOMBSTONE_BYTES = new byte[0];

  /**
   * A Helper class for performing functions common to all {@code DiskEntry}s.
   */
  class Helper {
    private static final Logger logger = LogService.getLogger();

    /**
     * Testing purpose only Get the value of an entry that is on disk without faulting it in and
     * without looking in the io buffer.
     *
     * @since GemFire 3.2.1
     */
    public static Object getValueOnDisk(DiskEntry entry, DiskRegion dr) {
      DiskId id = entry.getDiskId();
      if (id == null) {
        return null;
      }
      dr.acquireReadLock();
      try {
        synchronized (id) {
          if (dr.isBackup() && id.getKeyId() == DiskRegion.INVALID_ID || !entry.isValueNull() && id
              .needsToBeWritten() && !EntryBits
                  .isRecoveredFromDisk(id.getUserBits())/* fix for bug 41942 */) {
            return null;
          }

          return dr.getNoBuffer(id);
        }
      } finally {
        dr.releaseReadLock();
      }
    }

    /**
     * Get the serialized value directly from disk. Returned object may be a
     * {@link CachedDeserializable}. Goes straight to disk without faulting into memory. Only looks
     * at the disk storage, not at heap storage.
     *
     * @param entry the entry used to identify the value to fetch
     * @param dr the persistent storage from which to fetch the value
     * @return either null, byte array, or CacheDeserializable
     * @since GemFire 57_hotfix
     */
    public static Object getSerializedValueOnDisk(DiskEntry entry, DiskRegion dr) {
      DiskId did = entry.getDiskId();
      if (did == null) {
        return null;
      }
      dr.acquireReadLock();
      try {
        synchronized (did) {
          if (dr.isBackup() && did.getKeyId() == DiskRegion.INVALID_ID) {
            return null;
          } else if (!entry.isValueNull() && did.needsToBeWritten()
              && !EntryBits.isRecoveredFromDisk(did.getUserBits())/* fix for bug 41942 */) {
            return null;
          }
          return dr.getSerializedData(did);
        }
      } finally {
        dr.releaseReadLock();
      }
    }

    /**
     * Get the value of an entry that is on disk without faulting it in . It checks for the presence
     * in the buffer also. This method is used for concurrent map operations and CQ processing
     *
     * @since GemFire 5.1
     */
    public static Object getValueOnDiskOrBuffer(DiskEntry entry, DiskRegion dr,
        RegionEntryContext context) {
      @Released
      Object v = getOffHeapValueOnDiskOrBuffer(entry, dr, context);
      if (v instanceof CachedDeserializable) {
        CachedDeserializable cd = (CachedDeserializable) v;
        try {
          v = cd.getDeserializedValue(null, null);
        } finally {
          OffHeapHelper.release(cd); // If v was off-heap it is released here
        }
      }
      return v;
    }

    @Retained
    static Object getOffHeapValueOnDiskOrBuffer(DiskEntry entry, DiskRegion dr,
        RegionEntryContext context) {
      DiskId did = entry.getDiskId();
      Object syncObj = did;
      if (syncObj == null) {
        syncObj = entry;
      }
      if (syncObj == did) {
        dr.acquireReadLock();
      }
      try {
        synchronized (syncObj) {
          if (did != null && did.isPendingAsync()) {
            @Retained
            Object v = entry.getValueRetain(context, true);

            if (Token.isRemovedFromDisk(v)) {
              v = null;
            }
            return v;
          }
          if (did == null || (dr.isBackup() && did.getKeyId() == DiskRegion.INVALID_ID)
              || (!entry.isValueNull() && did.needsToBeWritten()
                  && !EntryBits.isRecoveredFromDisk(did.getUserBits()))/* fix for bug 41942 */) {
            return null;
          }

          return dr.getSerializedData(did);
        }
      } finally {
        if (syncObj == did) {
          dr.releaseReadLock();
        }
      }
    }

    static boolean isOverflowedToDisk(DiskEntry de, DiskRegion dr,
        DistributedRegion.DiskPosition dp, RegionEntryContext context) {
      Object v = null;
      DiskId did;
      synchronized (de) {
        did = de.getDiskId();
      }
      Object syncObj = did;
      if (syncObj == null) {
        syncObj = de;
      }
      if (syncObj == did) {
        dr.acquireReadLock();
      }
      try {
        synchronized (syncObj) {
          if (de.isValueNull()) {
            if (did == null) {
              synchronized (de) {
                did = de.getDiskId();
              }
              assert did != null;
              return isOverflowedToDisk(de, dr, dp, context);
            } else {
              dp.setPosition(did.getOplogId(), did.getOffsetInOplog());
              return true;
            }
          } else {
            return false;
          }
        }
      } finally {
        if (syncObj == did) {
          dr.releaseReadLock();
        }
      }
    }

    /**
     * Get the value of an entry that is on disk without faulting it in.
     *
     * @since GemFire 3.2.1
     */
    static boolean fillInValue(DiskEntry de, InitialImageOperation.Entry entry, DiskRegion dr,
        DistributionManager mgr, ByteArrayDataInput in, RegionEntryContext context,
        Version version) {
      @Retained
      @Released
      Object v = null;
      DiskId did;
      synchronized (de) {
        did = de.getDiskId();
      }
      Object syncObj = did;
      if (syncObj == null) {
        syncObj = de;
      }
      if (syncObj == did) {
        dr.acquireReadLock();
      }
      try {
        synchronized (syncObj) {
          entry.setLastModified(mgr, de.getLastModified());

          ReferenceCountHelper.setReferenceCountOwner(entry);

          // OFFHEAP copied to heap entry;
          // TODO: allow entry to refer to offheap since it will be copied to network.
          v = de.getValueRetain(context, true);

          ReferenceCountHelper.setReferenceCountOwner(null);
          if (v == null) {
            if (did == null) {
              // fix for bug 41449
              synchronized (de) {
                did = de.getDiskId();
              }
              assert did != null;
              // do recursive call to get readLock on did
              return fillInValue(de, entry, dr, mgr, in, context, version);
            }
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "DiskEntry.Helper.fillInValue, key={}; getting value from disk, disk id={}",
                  entry.getKey(), did);
            }
            BytesAndBits bb = null;
            try {
              bb = dr.getBytesAndBits(did, false);
            } catch (DiskAccessException ignore) {
              return false;
            }
            if (EntryBits.isInvalid(bb.getBits())) {
              entry.setInvalid();
            } else if (EntryBits.isLocalInvalid(bb.getBits())) {
              entry.setLocalInvalid();
            } else if (EntryBits.isTombstone(bb.getBits())) {
              entry.setTombstone();
            } else {
              entry.setValue(bb.getBytes());
              entry.setSerialized(EntryBits.isSerialized(bb.getBits()));
            }
            return true;
          }
        }
      } finally {
        if (syncObj == did) {
          dr.releaseReadLock();
        }
      }
      if (Token.isRemovedFromDisk(v)) {
        // fix for bug 31757
        return false;
      } else if (v instanceof CachedDeserializable) {
        CachedDeserializable cd = (CachedDeserializable) v;
        try {
          if (!cd.isSerialized()) {
            entry.setSerialized(false);
            entry.setValue(cd.getDeserializedForReading());

          } else {
            // don't serialize here if it is not already serialized

            Object tmp = cd.getValue();
            if (tmp instanceof byte[]) {
              entry.setValue((byte[]) tmp);
              entry.setSerialized(true);
            } else {
              try {
                HeapDataOutputStream hdos = new HeapDataOutputStream(version);
                BlobHelper.serializeTo(tmp, hdos);
                hdos.trim();
                entry.setValue(hdos);
                entry.setSerialized(true);
              } catch (IOException e) {
                throw new IllegalArgumentException(
                    "An IOException was thrown while serializing.",
                    e);
              }
            }
          }
        } finally {
          // If v == entry.value then v is assumed to be an OffHeapByteSource
          // and release() will be called on v after the bytes have been read from
          // off-heap.
          if (v != entry.getValue()) {
            OffHeapHelper.releaseWithNoTracking(v);
          }
        }
      } else if (v instanceof byte[]) {
        entry.setValue(v);
        entry.setSerialized(false);
      } else if (v == Token.INVALID) {
        entry.setInvalid();
      } else if (v == Token.LOCAL_INVALID) {
        // fix for bug 31107
        entry.setLocalInvalid();
      } else if (v == Token.TOMBSTONE) {
        entry.setTombstone();
      } else {
        Object preparedValue = v;
        if (preparedValue != null) {
          preparedValue = AbstractRegionEntry.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) {
            RuntimeException e2 = new IllegalArgumentException(
                "An IOException was thrown while serializing.");
            e2.initCause(e);
            throw e2;
          }
        }
      }
      return true;
    }

    /**
     * Used to initialize a new disk entry
     */
    public static void initialize(DiskEntry entry, DiskRecoveryStore diskRecoveryStore,
        Object newValue) {
      DiskRegionView drv = null;
      if (diskRecoveryStore instanceof InternalRegion) {
        drv = ((InternalRegion) diskRecoveryStore).getDiskRegion();
      } else if (diskRecoveryStore instanceof DiskRegionView) {
        drv = (DiskRegionView) diskRecoveryStore;
      }
      if (drv == null) {
        throw new IllegalArgumentException(
            "Disk region is null");
      }

      if (newValue instanceof RecoveredEntry) {
        // Set the id directly, the value will also be set if RECOVER_VALUES
        RecoveredEntry re = (RecoveredEntry) newValue;
        DiskId did = entry.getDiskId();
        did.setOplogId(re.getOplogId());
        did.setOffsetInOplog(re.getOffsetInOplog());
        did.setKeyId(re.getRecoveredKeyId());
        did.setUserBits(re.getUserBits());
        did.setValueLength(re.getValueLength());
        if (!re.getValueRecovered()) {
          updateStats(drv, diskRecoveryStore, 0, 1, did.getValueLength());
        } else {
          entry.setValueWithContext(drv, entry
              .prepareValueForCache((RegionEntryContext) diskRecoveryStore, re.getValue(), false));
          if (!Token.isInvalidOrRemoved(re.getValue())) {
            updateStats(drv, diskRecoveryStore, 1, 0, 0);
          }
        }
      } else {
        DiskId did = entry.getDiskId();
        if (did != null) {
          did.setKeyId(DiskRegion.INVALID_ID);
        }
        if (newValue != null && !Token.isInvalidOrRemoved(newValue)) {
          updateStats(drv, diskRecoveryStore, 1, 0, 0);
        }
      }
    }

    @Immutable
    private static final ValueWrapper INVALID_VW = new ByteArrayValueWrapper(true, INVALID_BYTES);
    @Immutable
    private static final ValueWrapper LOCAL_INVALID_VW =
        new ByteArrayValueWrapper(true, LOCAL_INVALID_BYTES);
    @Immutable
    private static final ValueWrapper TOMBSTONE_VW =
        new ByteArrayValueWrapper(true, TOMBSTONE_BYTES);

    public interface ValueWrapper {
      boolean isSerialized();

      int getLength();

      byte getUserBits();

      void sendTo(ByteBuffer bb, Flushable flushable) throws IOException;

      String getBytesAsString();
    }
    public interface Flushable {
      void flush() throws IOException;

      void flush(ByteBuffer bb, ByteBuffer chunkbb) throws IOException;
    }
    public static class ByteArrayValueWrapper implements ValueWrapper {
      public final boolean isSerializedObject;
      public final byte[] bytes;

      public ByteArrayValueWrapper(boolean isSerializedObject, byte[] bytes) {
        this.isSerializedObject = isSerializedObject;
        this.bytes = bytes;
      }

      @Override
      public boolean isSerialized() {
        return this.isSerializedObject;
      }

      @Override
      public int getLength() {
        return (this.bytes != null) ? this.bytes.length : 0;
      }

      private boolean isInvalidToken() {
        return this == INVALID_VW;
      }

      private boolean isLocalInvalidToken() {
        return this == LOCAL_INVALID_VW;
      }

      private boolean isTombstoneToken() {
        return this == TOMBSTONE_VW;
      }

      @Override
      public byte getUserBits() {
        byte userBits = 0x0;
        if (isSerialized()) {
          if (isTombstoneToken()) {
            userBits = EntryBits.setTombstone(userBits, true);
          } else if (isInvalidToken()) {
            userBits = EntryBits.setInvalid(userBits, true);
          } else if (isLocalInvalidToken()) {
            userBits = EntryBits.setLocalInvalid(userBits, true);
          } else {
            if (this.bytes == null) {
              throw new IllegalStateException("userBits==1 and value is null");
            } else if (this.bytes.length == 0) {
              throw new IllegalStateException("userBits==1 and value is zero length");
            }
            userBits = EntryBits.setSerialized(userBits, true);
          }
        }
        return userBits;
      }

      @Override
      public void sendTo(ByteBuffer bb, Flushable flushable) throws IOException {
        int offset = 0;
        final int maxOffset = getLength();
        while (offset < maxOffset) {
          int bytesThisTime = maxOffset - offset;
          boolean needsFlush = false;
          if (bytesThisTime > bb.remaining()) {
            needsFlush = true;
            bytesThisTime = bb.remaining();
          }
          bb.put(this.bytes, offset, bytesThisTime);
          offset += bytesThisTime;
          if (needsFlush) {
            flushable.flush();
          }
        }
      }

      @Override
      public String getBytesAsString() {
        if (this.bytes == null) {
          return "null";
        }
        StringBuilder sb = new StringBuilder();
        int len = getLength();
        for (int i = 0; i < len; i++) {
          sb.append(this.bytes[i]).append(", ");
        }
        return sb.toString();
      }
    }

    /**
     * This class is a bit of a hack used by the compactor. For the compactor always copies to a
     * byte[] so this class is just a simple wrapper. It is possible that the length of the byte
     * array is greater than the actual length of the wrapped data. At the time we create this we
     * are all done with isSerialized and userBits so those methods are not supported.
     */
    public static class CompactorValueWrapper extends ByteArrayValueWrapper {
      private final int length;

      public CompactorValueWrapper(byte[] bytes, int length) {
        super(false, bytes);
        this.length = length;
      }

      @Override
      public boolean isSerialized() {
        throw new UnsupportedOperationException();
      }

      @Override
      public int getLength() {
        return this.length;
      }

      @Override
      public byte getUserBits() {
        throw new UnsupportedOperationException();
      }
    }

    /**
     * Note that the StoredObject this ValueWrapper is created with is unretained so it must be used
     * before the owner of the StoredObject releases it. Since the RegionEntry that has the value we
     * are writing to disk has it retained we are ok as long as this ValueWrapper's life ends before
     * the RegionEntry sync is released. Note that this class is only used with uncompressed
     * StoredObjects.
     */
    public static class OffHeapValueWrapper implements ValueWrapper {
      private final @Unretained StoredObject offHeapData;

      public OffHeapValueWrapper(StoredObject so) {
        assert so.hasRefCount();
        assert !so.isCompressed();
        this.offHeapData = so;
      }

      @Override
      public boolean isSerialized() {
        return this.offHeapData.isSerialized();
      }

      @Override
      public int getLength() {
        return this.offHeapData.getDataSize();
      }

      @Override
      public byte getUserBits() {
        byte userBits = 0x0;
        if (isSerialized()) {
          userBits = EntryBits.setSerialized(userBits, true);
        }
        return userBits;
      }

      @Override
      public void sendTo(ByteBuffer bb, Flushable flushable) throws IOException {
        final int maxOffset = getLength();
        if (maxOffset == 0) {
          return;
        }
        if (maxOffset > bb.capacity()) {
          ByteBuffer chunkbb = this.offHeapData.createDirectByteBuffer();
          if (chunkbb != null) {
            flushable.flush(bb, chunkbb);
            return;
          }
        }
        final long bbAddress = AddressableMemoryManager.getDirectByteBufferAddress(bb);
        if (bbAddress != 0L) {
          int bytesRemaining = maxOffset;
          int availableSpace = bb.remaining();
          long addrToWrite = bbAddress + bb.position();
          long addrToRead = this.offHeapData.getAddressForReadingData(0, maxOffset);
          if (bytesRemaining > availableSpace) {
            do {
              AddressableMemoryManager.copyMemory(addrToRead, addrToWrite, availableSpace);
              bb.position(bb.position() + availableSpace);
              addrToRead += availableSpace;
              bytesRemaining -= availableSpace;
              flushable.flush();
              addrToWrite = bbAddress + bb.position();
              availableSpace = bb.remaining();
            } while (bytesRemaining > availableSpace);
          }
          AddressableMemoryManager.copyMemory(addrToRead, addrToWrite, bytesRemaining);
          bb.position(bb.position() + bytesRemaining);
        } else {
          long addr = this.offHeapData.getAddressForReadingData(0, maxOffset);
          final long endAddr = addr + maxOffset;
          while (addr != endAddr) {
            bb.put(AddressableMemoryManager.readByte(addr));
            addr++;
            if (!bb.hasRemaining()) {
              flushable.flush();
            }
          }
        }
      }

      @Override
      public String getBytesAsString() {
        return this.offHeapData.getStringForm();
      }
    }

    /**
     * Returns true if the given object is off-heap and it is worth wrapping a reference to it
     * instead of copying its data to the heap. Currently all StoredObject's with a refCount are
     * wrapped.
     */
    public static boolean wrapOffHeapReference(Object o) {
      if (o instanceof StoredObject) {
        StoredObject so = (StoredObject) o;
        if (so.hasRefCount()) {
          //
          return true;
        }
      }
      return false;
    }

    public static ValueWrapper createValueWrapper(Object value, EntryEventImpl event) {
      if (value == Token.INVALID) {
        // even though it is not serialized we say it is because
        // bytes will never be an empty array when it is serialized
        // so that gives us a way to specify the invalid value
        // given a byte array and a boolean flag.
        return INVALID_VW;
      } else if (value == Token.LOCAL_INVALID) {
        // even though it is not serialized we say it is because
        // bytes will never be an empty array when it is serialized
        // so that gives us a way to specify the local-invalid value
        // given a byte array and a boolean flag.
        return LOCAL_INVALID_VW;
      } else if (value == Token.TOMBSTONE) {
        return TOMBSTONE_VW;
      } else {
        boolean isSerializedObject = true;
        byte[] bytes;
        if (value instanceof CachedDeserializable) {
          CachedDeserializable proxy = (CachedDeserializable) value;
          if (wrapOffHeapReference(proxy)) {
            return new OffHeapValueWrapper((StoredObject) proxy);
          }
          isSerializedObject = proxy.isSerialized();
          if (isSerializedObject) {
            bytes = proxy.getSerializedValue();
          } else {
            bytes = (byte[]) proxy.getDeserializedForReading();
          }
          if (event != null && isSerializedObject) {
            event.setCachedSerializedNewValue(bytes);
          }
        } else if (value instanceof byte[]) {
          isSerializedObject = false;
          bytes = (byte[]) value;
        } else {
          Assert.assertTrue(!Token.isRemovedFromDisk(value));
          if (event != null && event.getCachedSerializedNewValue() != null) {
            bytes = event.getCachedSerializedNewValue();
          } else {
            bytes = EntryEventImpl.serialize(value);
            if (bytes.length == 0) {
              throw new IllegalStateException(
                  "serializing <" + value + "> produced empty byte array");
            }
            if (event != null) {
              event.setCachedSerializedNewValue(bytes);
            }
          }
        }
        return new ByteArrayValueWrapper(isSerializedObject, bytes);
      }
    }

    public static ValueWrapper createValueWrapperFromEntry(DiskEntry entry, InternalRegion region,
        EntryEventImpl event) {
      if (event != null) {
        // For off-heap it should be faster to pass a reference to the
        // StoredObject instead of using the cached byte[] (unless it is also compressed).
        // Since NIO is used if the chunk of memory is large we can write it
        // to the file using the off-heap memory with no extra copying.
        // So we give preference to getRawNewValue over getCachedSerializedNewValue
        Object rawValue = null;
        {
          rawValue = event.getRawNewValue();
          if (wrapOffHeapReference(rawValue)) {
            return new OffHeapValueWrapper((StoredObject) rawValue);
          }
        }
        if (event.getCachedSerializedNewValue() != null) {
          return new ByteArrayValueWrapper(true, event.getCachedSerializedNewValue());
        }
        if (rawValue != null) {
          return createValueWrapper(rawValue, event);
        }
      }
      @Retained
      Object value = entry.getValueRetain(region, true);
      try {
        return createValueWrapper(value, event);
      } finally {
        OffHeapHelper.release(value);
      }
    }

    private static void writeToDisk(DiskEntry entry, InternalRegion region, boolean async)
        throws RegionClearedException {
      writeToDisk(entry, region, async, null);
    }

    /**
     * Writes the key/value object stored in the given entry to disk
     *
     * @see DiskRegion#put
     */
    private static void writeToDisk(DiskEntry entry, InternalRegion region, boolean async,
        EntryEventImpl event) throws RegionClearedException {
      writeBytesToDisk(entry, region, async, createValueWrapperFromEntry(entry, region, event));
    }

    private static void writeBytesToDisk(DiskEntry entry, InternalRegion region, boolean async,
        ValueWrapper vw) throws RegionClearedException {
      // @todo does the following unmark need to be called when an async
      // write is scheduled or is it ok for doAsyncFlush to do it?
      entry.getDiskId().unmarkForWriting();
      region.getDiskRegion().put(entry, region, vw, async);
    }

    public static void update(DiskEntry entry, InternalRegion region, @Unretained Object newValue)
        throws RegionClearedException {
      update(entry, region, newValue, null);
    }

    /**
     * Updates the value of the disk entry with a new value. This allows us to free up disk space in
     * the non-backup case.
     */
    public static void update(DiskEntry entry, InternalRegion region, @Unretained Object newValue,
        EntryEventImpl event) throws RegionClearedException {
      if (newValue == null) {
        throw new NullPointerException(
            "Entry's value should not be null.");
      }
      boolean basicUpdateCalled = false;
      try {

        AsyncDiskEntry asyncDiskEntry = null;
        DiskRegion dr = region.getDiskRegion();
        DiskId did = entry.getDiskId();
        Object syncObj = did;
        if (syncObj == null) {
          syncObj = entry;
        }
        if (syncObj == did) {
          dr.acquireReadLock();
        }
        try {
          synchronized (syncObj) {
            basicUpdateCalled = true;
            asyncDiskEntry = basicUpdate(entry, region, newValue, event);
          }
        } finally {
          if (syncObj == did) {
            dr.releaseReadLock();
          }
        }
        if (asyncDiskEntry != null && did.isPendingAsync()) {
          // this needs to be done outside the above sync
          scheduleAsyncWrite(asyncDiskEntry);
        }
      } finally {
        if (!basicUpdateCalled) {
          OffHeapHelper.release(newValue);
        }
      }
    }

    static AsyncDiskEntry basicUpdateForTesting(DiskEntry entry, InternalRegion region,
        @Unretained Object newValue, EntryEventImpl event) throws RegionClearedException {
      return basicUpdate(entry, region, newValue, event);
    }

    private static AsyncDiskEntry basicUpdate(DiskEntry entry, InternalRegion region,
        @Unretained Object newValue, EntryEventImpl event) throws RegionClearedException {
      AsyncDiskEntry result = null;
      DiskRegion dr = region.getDiskRegion();
      DiskId did = entry.getDiskId();
      Token oldValue;
      int oldValueLength;
      oldValue = entry.getValueAsToken();
      if (Token.isRemovedFromDisk(newValue)) {
        if (dr.isBackup()) {
          dr.testIsRecoveredAndClear(did); // fixes bug 41409
        }
        boolean caughtCacheClosed = false;
        try {
          if (!Token.isRemovedFromDisk(oldValue)) {
            result = basicRemoveFromDisk(entry, region, false);
          }
        } catch (CacheClosedException e) {
          caughtCacheClosed = true;
          throw e;
        } finally {
          if (caughtCacheClosed) {
            // 47616: not to set the value to be removedFromDisk since it failed to persist
          } else {
            // Ensure that the value is rightly set despite clear so
            // that it can be distributed correctly
            entry.setValueWithContext(region, newValue); // OFFHEAP newValue was already
                                                         // preparedForCache
          }
        }
      } else if (newValue instanceof RecoveredEntry) {
        ((RecoveredEntry) newValue).applyToDiskEntry(entry, region, dr, did);
      } else {
        boolean newValueStoredInEntry = false;
        try {
          // The new value in the entry needs to be set after the disk writing
          // has succeeded.

          // entry.setValueWithContext(region, newValue); // OFFHEAP newValue already prepared

          if (did != null && did.isPendingAsync()) {
            // if the entry was not yet written to disk, we didn't update
            // the bytes on disk.
            oldValueLength = 0;
          } else {
            oldValueLength = getValueLength(did);
          }

          if (dr.isBackup()) {
            dr.testIsRecoveredAndClear(did); // fixes bug 41409
            if (doSynchronousWrite(region, dr)) {
              if (AbstractRegionEntry.isCompressible(dr, newValue)) {
                // In case of compression the value is being set first
                // so that writeToDisk can get it back from the entry
                // decompressed if it does not have it already in the event.
                // TODO: this may have introduced a bug with clear since
                // writeToDisk can throw RegionClearedException which
                // was supposed to stop us from changing entry.
                newValueStoredInEntry = true;
                entry.setValueWithContext(region, newValue); // OFFHEAP newValue already prepared
                // newValue is prepared and compressed. We can't write compressed values to disk.
                if (!entry.isRemovedFromDisk()) {
                  writeToDisk(entry, region, false, event);
                }
              } else {
                writeBytesToDisk(entry, region, false, createValueWrapper(newValue, event));
                newValueStoredInEntry = true;
                entry.setValueWithContext(region, newValue); // OFFHEAP newValue already prepared
              }

            } else {
              // If we have concurrency checks enabled for a persistent region, we need
              // to add an entry to the async queue for every update to maintain the RVV
              boolean maintainRVV = region.getConcurrencyChecksEnabled();

              if (!did.isPendingAsync() || maintainRVV) {
                // if the entry is not async, we need to schedule it
                // for regions with concurrency checks enabled, we add an entry
                // to the queue for every entry.
                did.setPendingAsync(true);
                VersionTag tag = null;
                VersionStamp stamp = entry.getVersionStamp();
                if (stamp != null) {
                  tag = stamp.asVersionTag();
                }
                result = new AsyncDiskEntry(region, entry, tag);
              }
              newValueStoredInEntry = true;
              entry.setValueWithContext(region, newValue); // OFFHEAP newValue already prepared
            }
          } else if (did != null) {
            newValueStoredInEntry = true;
            entry.setValueWithContext(region, newValue); // OFFHEAP newValue already prepared

            // Mark the id as needing to be written
            // The disk remove that this section used to do caused bug 30961
            // @todo this seems wrong. How does leaving it on disk fix the bug?
            did.markForWriting();
            // did.setValueSerializedSize(0);
          } else {
            newValueStoredInEntry = true;
            entry.setValueWithContext(region, newValue);
          }
        } finally {
          if (!newValueStoredInEntry) {
            OffHeapHelper.release(newValue);
          }
        }


        if (Token.isInvalidOrRemoved(newValue)) {
          if (oldValue == null) {
            updateStats(dr, region, 0/* InVM */, -1/* OnDisk */, -oldValueLength);
          } else if (!Token.isInvalidOrRemoved(oldValue)) {
            updateStats(dr, region, -1/* InVM */, 0/* OnDisk */, 0);
          } else {
            // oldValue was also a token which is neither in vm or on disk.
          }
        } else { // we have a value to put in the vm
          if (oldValue == null) {
            updateStats(dr, region, 1/* InVM */, -1/* OnDisk */, -oldValueLength);
          } else if (Token.isInvalidOrRemoved(oldValue)) {
            updateStats(dr, region, 1/* InVM */, 0/* OnDisk */, 0/* overflowBytesOnDisk */);
          } else {
            // old value was also in vm so no change
          }
        }
      }
      if (entry instanceof EvictableEntry) {
        EvictableEntry le = (EvictableEntry) entry;
        le.unsetEvicted();
      }
      return result;
    }

    private static int getValueLength(DiskId did) {
      int result = 0;
      if (did != null) {
        synchronized (did) {
          result = did.getValueLength();
        }
      }
      return result;
    }

    public static Object getValueInVMOrDiskWithoutFaultIn(DiskEntry entry, InternalRegion region) {
      Object result = OffHeapHelper.copyAndReleaseIfNeeded(
          getValueOffHeapOrDiskWithoutFaultIn(entry, region), region.getCache());
      if (result instanceof CachedDeserializable) {
        result = ((CachedDeserializable) result).getDeserializedValue(null, null);
      }
      return result;
    }

    @Retained
    public static Object getValueOffHeapOrDiskWithoutFaultIn(DiskEntry entry,
        InternalRegion region) {
      @Retained
      Object v = entry.getValueRetain(region, true);

      if (v == null || Token.isRemovedFromDisk(v) && !region.isIndexCreationThread()) {
        synchronized (entry) {
          v = entry.getValueRetain(region, true);

          if (v == null) {
            v = Helper.getOffHeapValueOnDiskOrBuffer(entry, region.getDiskRegion(), region);
          }
        }
      }
      if (Token.isRemovedFromDisk(v)) {
        // fix for bug 31800
        v = null;
      }
      return v;
    }

    public static Object faultInValue(DiskEntry entry, InternalRegion region) {
      return faultInValue(entry, region, false);
    }

    @Retained
    public static Object faultInValueRetain(DiskEntry entry, InternalRegion region) {
      return faultInValue(entry, region, true);
    }

    /**
     * @param retainResult if true then the result may be a retained off-heap reference
     */
    @Retained
    private static Object faultInValue(DiskEntry entry, InternalRegion region,
        boolean retainResult) {
      DiskRegion dr = region.getDiskRegion();
      @Retained
      Object v = entry.getValueRetain(region, true);

      boolean lruFaultedIn = false;
      boolean done = false;
      try {
        if (entry instanceof EvictableEntry && !dr.isSync()) {
          synchronized (entry) {
            DiskId did = entry.getDiskId();
            if (did != null && did.isPendingAsync()) {
              done = true;
              // See if it is pending async because of a faultOut.
              // If so then if we are not a backup then we can unschedule the pending async.
              // In either case we need to do the lruFaultIn logic.
              boolean evicted = ((EvictableEntry) entry).isEvicted();
              if (evicted) {
                if (!dr.isBackup()) {
                  // @todo do we also need a bit that tells us if it is in the async queue?
                  // Seems like we could end up adding it to the queue multiple times.
                  did.setPendingAsync(false);
                }
              }
              lruEntryFaultIn((EvictableEntry) entry, (DiskRecoveryStore) region);
              lruFaultedIn = true;
            }
          }
        }
        if (!done && (v == null || Token.isRemovedFromDisk(v) && !region.isIndexCreationThread())) {
          synchronized (entry) {
            v = entry.getValueRetain(region, true);

            if (v == null) {
              v = readValueFromDisk(entry, (DiskRecoveryStore) region);
              if (entry instanceof EvictableEntry) {
                if (v != null && !Token.isInvalid(v)) {
                  lruEntryFaultIn((EvictableEntry) entry, (DiskRecoveryStore) region);

                  lruFaultedIn = true;
                }
              }
            }
          }
        }
      } finally {
        if (!retainResult) {
          v = OffHeapHelper.copyAndReleaseIfNeeded(v, region.getCache());
          // At this point v should be either a heap object
        }
      }
      if (Token.isRemoved(v)) {
        // fix for bug 31800
        v = null;
      } else {
        entry.setRecentlyUsed(region);
      }
      if (lruFaultedIn) {
        lruUpdateCallback((DiskRecoveryStore) region);
      }
      return v; // OFFHEAP: the value ends up being returned by RegionEntry.getValue
    }

    public static void recoverValue(DiskEntry entry, long oplogId, DiskRecoveryStore recoveryStore,
        ByteArrayDataInput in) {
      boolean lruFaultedIn = false;
      synchronized (entry) {
        if (entry.isValueNull()) {
          DiskId did = entry.getDiskId();
          if (did != null) {
            Object value = null;
            DiskRegionView dr = recoveryStore.getDiskRegionView();
            dr.acquireReadLock();
            try {
              synchronized (did) {
                // don't read if the oplog has changed.
                if (oplogId == did.getOplogId()) {
                  value = getValueFromDisk(dr, did, in, dr.getCache());
                  if (value != null) {
                    setValueOnFaultIn(value, did, entry, dr, recoveryStore);
                  }
                }
              }
            } finally {
              dr.releaseReadLock();
            }
            if (entry instanceof EvictableEntry) {
              if (value != null && !Token.isInvalid(value)) {
                lruEntryFaultIn((EvictableEntry) entry, recoveryStore);
                lruFaultedIn = true;
              }
            }
          }
        }
      }
      if (lruFaultedIn) {
        lruUpdateCallback(recoveryStore);
      }
    }

    /**
     * Caller must have "did" synced.
     */
    private static Object getValueFromDisk(DiskRegionView dr, DiskId did, ByteArrayDataInput in,
        InternalCache cache) {
      Object value;
      if (dr.isBackup() && did.getKeyId() == DiskRegion.INVALID_ID) {
        // must have been destroyed
        value = null;
      } else {
        // if a bucket region then create a CachedDeserializable here instead of object
        value = dr.getRaw(did); // fix bug 40192
        if (value instanceof BytesAndBits) {
          BytesAndBits bb = (BytesAndBits) value;
          if (EntryBits.isInvalid(bb.getBits())) {
            value = Token.INVALID;
          } else if (EntryBits.isLocalInvalid(bb.getBits())) {
            value = Token.LOCAL_INVALID;
          } else if (EntryBits.isTombstone(bb.getBits())) {
            value = Token.TOMBSTONE;
          } else if (EntryBits.isSerialized(bb.getBits())) {
            value = readSerializedValue(bb.getBytes(), bb.getVersion(), in, false, cache);
          } else {
            value = readRawValue(bb.getBytes(), bb.getVersion(), in);
          }
        }
      }
      return value;
    }

    private static void lruUpdateCallback(DiskRecoveryStore recoveryStore) {
      /*
       * Used conditional check to see if if its a LIFO Enabled, yes then disable
       * lruUpdateCallback() and called updateStats() its keep track of actual entries present in
       * memory - useful when checking capacity constraint
       */
      try {
        if (recoveryStore.getEvictionAttributes() != null
            && recoveryStore.getEvictionAttributes().getAlgorithm().isLIFO()) {
          recoveryStore.getRegionMap().updateEvictionCounter();
          return;
        }
        // this must be done after releasing synchronization
        recoveryStore.getRegionMap().lruUpdateCallback();
      } catch (DiskAccessException dae) {
        recoveryStore.handleDiskAccessException(dae);
        throw dae;
      }
    }

    private static void lruEntryFaultIn(EvictableEntry entry, DiskRecoveryStore recoveryStore) {
      RegionMap rm = (RegionMap) recoveryStore.getRegionMap();
      try {
        rm.lruEntryFaultIn((EvictableEntry) entry);
      } catch (DiskAccessException dae) {
        recoveryStore.handleDiskAccessException(dae);
        throw dae;
      }
    }

    /**
     * Returns the value of this map entry, reading it from disk, if necessary. Sets the value in
     * the entry. This is only called by the faultIn code once it has determined that the value is
     * no longer in memory. Caller must have "entry" synced.
     */
    private static Object readValueFromDisk(DiskEntry entry, DiskRecoveryStore region) {

      DiskRegionView dr = region.getDiskRegionView();
      DiskId did = entry.getDiskId();
      if (did == null) {
        return null;
      }
      dr.acquireReadLock();
      try {
        synchronized (did) {
          Object value = getValueFromDisk(dr, did, null, dr.getCache());
          if (value == null)
            return null;
          setValueOnFaultIn(value, did, entry, dr, region);
          return value;
        }
      } finally {
        dr.releaseReadLock();
      }
    }

    /**
     * Caller must have "entry" and "did" synced and "dr" readLocked.
     *
     * @return the unretained result must be used by the caller before it releases the sync on
     *         "entry".
     */
    @Unretained
    private static Object setValueOnFaultIn(Object value, DiskId did, DiskEntry entry,
        DiskRegionView dr, DiskRecoveryStore region) {
      // dr.getOwner().getCache().getLogger().info("DEBUG: faulting in entry with key " +
      // entry.getKey());
      int bytesOnDisk = getValueLength(did);
      // Retained by the prepareValueForCache call for the region entry.
      // NOTE that we return this value unretained because the retain is owned by the region entry
      // not the caller.
      @Retained
      Object preparedValue = entry.prepareValueForCache((RegionEntryContext) region, value, false);
      region.updateSizeOnFaultIn(entry.getKey(), region.calculateValueSize(preparedValue),
          bytesOnDisk);
      // did.setValueSerializedSize(0);
      // I think the following assertion is true but need to run
      // a regression with it. Reenable this post 6.5
      // Assert.assertTrue(entry._getValue() == null);
      entry.setValueWithContext((RegionEntryContext) region, preparedValue);
      if (!Token.isInvalidOrRemoved(value)) {
        updateStats(dr, region, 1/* InVM */, -1/* OnDisk */, -bytesOnDisk);
      }
      return preparedValue;
    }

    public static Object readSerializedValue(byte[] valueBytes, Version version,
        ByteArrayDataInput in, boolean forceDeserialize, InternalCache cache) {
      if (forceDeserialize) {
        // deserialize checking for product version change
        return EntryEventImpl.deserialize(valueBytes, version, in);
      } else {
        // TODO: upgrades: is there a case where GemFire values are internal
        // ones that need to be upgraded transparently; probably messages
        // being persisted (gateway events?)
        return CachedDeserializableFactory.create(valueBytes, cache);
      }
    }

    public static Object readRawValue(byte[] valueBytes, Version version, ByteArrayDataInput in) {
      return valueBytes;
    }

    public static void updateStats(DiskRegionView drv, Object owner, int entriesInVmDelta,
        int overflowOnDiskDelta, int overflowBytesOnDiskDelta) {
      if (entriesInVmDelta != 0) {
        drv.incNumEntriesInVM(entriesInVmDelta);
      }
      if (overflowOnDiskDelta != 0) {
        drv.incNumOverflowOnDisk(overflowOnDiskDelta);
      }
      if (overflowBytesOnDiskDelta != 0) {
        drv.incNumOverflowBytesOnDisk(overflowBytesOnDiskDelta);
      }
      if (owner instanceof BucketRegion) {
        BucketRegion br = (BucketRegion) owner;
        br.incNumEntriesInVM(entriesInVmDelta);
        br.incNumOverflowOnDisk(overflowOnDiskDelta);
        br.incNumOverflowBytesOnDisk(overflowBytesOnDiskDelta);
      }
      // Note: we used to call owner.incNumOverflowBytesOnDisk()
      // if owner was a DiskRegionView.
      // But since we also call drv.incNumOverflowBytesOnDisk()
      // and since drv is == owner when owner is not a InternalRegion
      // (see PlaceHolderDiskRegion.getDiskRegionView())
      // this resulted in incNumOverflowBytesOnDisk being called twice.
    }

    /**
     * Writes the value of this {@code DiskEntry} to disk and {@code null} s out the reference to
     * the value to free up VM space.
     * <p>
     * Note that if the value had already been written to disk, it is not written again.
     * <p>
     * Caller must synchronize on entry and it is assumed the entry is evicted
     */
    public static int overflowToDisk(DiskEntry entry, InternalRegion region,
        EvictionController ccHelper) throws RegionClearedException {
      DiskRegion dr = region.getDiskRegion();
      final int oldSize = region.calculateRegionEntryValueSize(entry);
      // Get diskID . If it is null, it implies it is overflow only mode.
      DiskId did = entry.getDiskId();
      if (did == null) {
        ((EvictableEntry) entry).setDelayedDiskId((DiskRecoveryStore) region);
        did = entry.getDiskId();
      }

      int change = 0;
      boolean scheduledAsyncHere = false;
      dr.acquireReadLock();
      try {
        synchronized (did) {
          // check for a concurrent freeAllEntriesOnDisk which syncs on DiskId but not on the entry
          if (entry.isRemovedFromDisk()) {
            return 0;
          }

          // TODO: Check if we need to overflow even when id is = 0
          boolean wasAlreadyPendingAsync = did.isPendingAsync();
          if (did.needsToBeWritten()) {
            if (doSynchronousWrite(region, dr)) {
              writeToDisk(entry, region, false);
            } else if (!wasAlreadyPendingAsync) {
              scheduledAsyncHere = true;
              did.setPendingAsync(true);
            } else {
              // it may have been scheduled to be written (isBackup==true)
              // and now we are faulting it out
            }
          }

          // If async then if it does not need to be written (because it already was)
          // then treat it like the sync case. This fixes bug 41310
          if (scheduledAsyncHere || wasAlreadyPendingAsync) {
            // we call _setValue(null) after it is actually written to disk
            change = entry.updateAsyncEntrySize(ccHelper);
            // do the stats when it is actually written to disk
          } else {
            region.updateSizeOnEvict(entry.getKey(), oldSize);
            entry.handleValueOverflow(region);
            entry.setValueWithContext(region, null);
            change = ((EvictableEntry) entry).updateEntrySize(ccHelper);
            // the caller checked to make sure we had something to overflow
            // so dec inVM and inc onDisk
            updateStats(dr, region, -1/* InVM */, 1/* OnDisk */, getValueLength(did));
          }
        }
      } finally {
        dr.releaseReadLock();
      }
      if (scheduledAsyncHere && did.isPendingAsync()) {
        // this needs to be done outside the above sync
        // the version tag is null here because this method only needs
        // to write to disk for overflow only regions, which do not need
        // to maintain an RVV on disk.
        scheduleAsyncWrite(new AsyncDiskEntry(region, entry, null));
      }
      return change;
    }

    private static void scheduleAsyncWrite(AsyncDiskEntry ade) {
      DiskRegion dr = ade.region.getDiskRegion();
      dr.scheduleAsyncWrite(ade);
    }

    public static void handleFullAsyncQueue(DiskEntry entry, InternalRegion region,
        VersionTag tag) {
      writeEntryToDisk(entry, region, tag, true);
    }

    public static void doAsyncFlush(VersionTag tag, InternalRegion region) {
      if (region.isThisRegionBeingClosedOrDestroyed())
        return;
      DiskRegion dr = region.getDiskRegion();
      if (!dr.isBackup()) {
        return;
      }
      assert !dr.isSync();
      dr.acquireReadLock();
      try {
        dr.getDiskStore().putVersionTagOnly(region, tag, true);
      } finally {
        dr.releaseReadLock();
      }
    }

    /**
     * Flush an entry that was previously scheduled to be written to disk.
     *
     * @since GemFire prPersistSprint1
     */
    public static void doAsyncFlush(DiskEntry entry, InternalRegion region, VersionTag tag) {
      writeEntryToDisk(entry, region, tag, false);
    }

    /**
     * Does a synchronous write to disk for a region that uses async. This method is used by both
     * doAsyncFlush and handleFullAsyncQueue to fix GEODE-1700.
     *
     * @param asyncQueueWasFull true if caller wanted to put this entry in the queue but could not
     *        do so because it was full
     */
    private static void writeEntryToDisk(DiskEntry entry, InternalRegion region, VersionTag tag,
        boolean asyncQueueWasFull) {
      if (region.isThisRegionBeingClosedOrDestroyed())
        return;
      DiskRegion dr = region.getDiskRegion();
      if (!asyncQueueWasFull) {
        dr.setClearCountReference();
      }
      synchronized (entry) { // fixes 40116
        // If I don't sync the entry and this method ends up doing an eviction
        // thus setting value to null
        // some other thread is free to fetch the value while the entry is synced
        // and think it has removed it or replaced it. This results in updateSizeOn*
        // being called twice for the same value (once when it is evicted and once
        // when it is removed/updated).
        try {
          dr.acquireReadLock();
          try {
            DiskId did = entry.getDiskId();
            synchronized (did) {
              if (did.isPendingAsync()) {
                did.setPendingAsync(false);
                final Token entryVal = entry.getValueAsToken();
                final int entryValSize = region.calculateRegionEntryValueSize(entry);
                try {
                  if (Token.isRemovedFromDisk(entryVal)) {
                    if (region.isThisRegionBeingClosedOrDestroyed())
                      return;
                    dr.remove(region, entry, true, false);
                    if (dr.isBackup()) {
                      did.setKeyId(DiskRegion.INVALID_ID); // fix for bug 41340
                    }
                  } else if ((Token.isInvalid(entryVal) || entryVal == Token.TOMBSTONE)
                      && !dr.isBackup()) {
                    // no need to write invalid or tombstones to disk if overflow only
                  } else if (entryVal != null) {
                    writeToDisk(entry, region, true);
                    assert !dr.isSync();
                    // Only setValue to null if this was an evict.
                    // We could just be a backup that is writing async.
                    if (!Token.isInvalid(entryVal) && (entryVal != Token.TOMBSTONE)
                        && entry instanceof EvictableEntry
                        && ((EvictableEntry) entry).isEvicted()) {
                      // Moved this here to fix bug 40116.
                      region.updateSizeOnEvict(entry.getKey(), entryValSize);
                      updateStats(dr, region, -1/* InVM */, 1/* OnDisk */, did.getValueLength());
                      entry.handleValueOverflow(region);
                      entry.setValueWithContext(region, null);
                    }
                  } else {
                    // if we have a version tag we need to record the operation
                    // to update the RVV
                    if (tag != null) {
                      DiskEntry.Helper.doAsyncFlush(tag, region);
                    }
                    return;
                  }
                } catch (RegionClearedException ignore) {
                  // no need to do the op since it was clobbered by a region clear
                }

                // See if we the entry we wrote to disk has the same tag
                // as this entry. If not, write the tag as a conflicting operation.
                // to update the RVV.
                VersionStamp stamp = entry.getVersionStamp();
                if (tag != null && stamp != null && (stamp.getMemberID() != tag.getMemberID()
                    || stamp.getRegionVersion() != tag.getRegionVersion())) {
                  DiskEntry.Helper.doAsyncFlush(tag, region);
                }
              } else {
                // if we have a version tag we need to record the operation
                // to update the RVV
                if (tag != null) {
                  DiskEntry.Helper.doAsyncFlush(tag, region);
                }
              }
            }
          } finally {
            dr.releaseReadLock();
          }
        } finally {
          if (!asyncQueueWasFull) {
            dr.removeClearCountReference();
          }
        }
      } // sync entry
    }

    /**
     * Removes the key/value pair in the given entry from disk
     *
     * @throws RegionClearedException If the operation is aborted due to a clear
     * @see DiskRegion#remove
     */
    public static void removeFromDisk(DiskEntry entry, InternalRegion region, boolean isClear)
        throws RegionClearedException {
      DiskRegion dr = region.getDiskRegion();
      DiskId did = entry.getDiskId();
      Object syncObj = did;
      if (did == null) {
        syncObj = entry;
      }
      AsyncDiskEntry asyncDiskEntry = null;
      if (syncObj == did) {
        dr.acquireReadLock();
      }
      try {
        synchronized (syncObj) {
          asyncDiskEntry = basicRemoveFromDisk(entry, region, isClear);
        }
      } finally {
        if (syncObj == did) {
          dr.releaseReadLock();
        }
      }
      if (asyncDiskEntry != null && did.isPendingAsync()) {
        // do this outside the sync
        scheduleAsyncWrite(asyncDiskEntry);
      }
    }

    private static AsyncDiskEntry basicRemoveFromDisk(DiskEntry entry, InternalRegion region,
        boolean isClear) throws RegionClearedException {
      final DiskRegion dr = region.getDiskRegion();
      final DiskId did = entry.getDiskId();
      final Object curValAsToken = entry.getValueAsToken();
      if (did == null || (dr.isBackup() && did.getKeyId() == DiskRegion.INVALID_ID)) {
        // Not on disk yet
        if (!Token.isInvalidOrRemoved(curValAsToken)) {
          updateStats(dr, region, -1/* InVM */, 0/* OnDisk */, 0);
        }
        dr.unscheduleAsyncWrite(did);
        return null;
      }
      AsyncDiskEntry result = null;

      // Bug # 39989
      did.unmarkForWriting();

      // System.out.println("DEBUG: removeFromDisk doing remove(" + id + ")");
      int oldValueLength = did.getValueLength();
      if (doSynchronousWrite(region, dr) || isClear) {
        dr.remove(region, entry, false, isClear);
        if (dr.isBackup()) {
          did.setKeyId(DiskRegion.INVALID_ID); // fix for bug 41340
        }
        // If this is a clear, we should unschedule the async write for this
        // entry
        did.setPendingAsync(false);
      } else {
        // If we have concurrency checks enabled for a persistent region, we need
        // to add an entry to the async queue for every update to maintain the RVV
        boolean maintainRVV = region.getConcurrencyChecksEnabled() && dr.isBackup();
        if (!did.isPendingAsync() || maintainRVV) {
          did.setPendingAsync(true);
          VersionTag tag = null;
          VersionStamp stamp = entry.getVersionStamp();
          if (stamp != null) {
            tag = stamp.asVersionTag();
          }
          result = new AsyncDiskEntry(region, entry, tag);
        }
      }
      if (curValAsToken == null) {
        updateStats(dr, region, 0/* InVM */, -1/* OnDisk */, -oldValueLength);
      } else if (!Token.isInvalidOrRemoved(curValAsToken)) {
        updateStats(dr, region, -1/* InVM */, 0/* OnDisk */, 0);
      }
      return result;
    }

    public static void updateVersionOnly(DiskEntry entry, InternalRegion region, VersionTag tag) {
      DiskRegion dr = region.getDiskRegion();
      if (!dr.isBackup()) {
        return;
      }

      assert tag != null && tag.getMemberID() != null;
      boolean scheduleAsync = false;
      DiskId did = entry.getDiskId();
      Object syncObj = did;
      if (syncObj == null) {
        syncObj = entry;
      }
      if (syncObj == did) {
        dr.acquireReadLock();
      }
      try {
        synchronized (syncObj) {
          if (doSynchronousWrite(region, dr)) {
            dr.getDiskStore().putVersionTagOnly(region, tag, false);
          } else {
            scheduleAsync = true;
          }
        }
      } finally {
        if (syncObj == did) {
          dr.releaseReadLock();
        }
      }
      if (scheduleAsync) {
        // this needs to be done outside the above sync
        scheduleAsyncWrite(new AsyncDiskEntry(region, tag));
      }
    }

    static boolean doSynchronousWrite(InternalRegion region, DiskRegion dr) {
      return dr.isSync() || (dr.isBackup() && !region.isInitialized());
    }

  }

  /**
   * A marker object for an entry that has been recovered from disk. It is handled specially when it
   * is placed in a region.
   */
  class RecoveredEntry {

    /** The disk id of the entry being recovered */
    private final long recoveredKeyId;

    /** The value of the recovered entry */
    private final Object value;

    private final long offsetInOplog;
    private final byte userBits;
    private final int valueLength;

    private long oplogId;
    private VersionTag tag;

    /** whether the entry value has been faulted in after recovery. */
    private final boolean valueRecovered;

    /**
     * Only for this constructor, the value is not loaded into the region & it is lying on the
     * oplogs. Since Oplogs rely on DiskId to furnish user bits so as to correctly interpret bytes,
     * the userbit needs to be set correctly here.
     */
    public RecoveredEntry(long keyId, long oplogId, long offsetInOplog, byte userBits,
        int valueLength) {
      this(keyId, oplogId, offsetInOplog, userBits, valueLength, null, false);
    }

    public RecoveredEntry(long keyId, long oplogId, long offsetInOplog, byte userBits,
        int valueLength, Object value) {
      this(keyId, oplogId, offsetInOplog, userBits, valueLength, value, true);
    }

    public RecoveredEntry(long keyId, long oplogId, long offsetInOplog, byte userBits,
        int valueLength, Object value, boolean valueRecovered) {
      this.recoveredKeyId = keyId;
      this.value = value;
      this.oplogId = oplogId;
      this.offsetInOplog = offsetInOplog;
      this.userBits = EntryBits.setRecoveredFromDisk(userBits, true);
      this.valueLength = valueLength;
      this.valueRecovered = valueRecovered;
    }

    /**
     * Returns the disk id of the entry being recovered
     */
    public long getRecoveredKeyId() {
      return this.recoveredKeyId;
    }

    /**
     * Returns the value of the recovered entry. Note that if the disk id is < 0 then the value has
     * not been faulted in and this method will return null.
     */
    public Object getValue() {
      return this.value;
    }

    /**
     * @return byte indicating the user bits. The correct value is returned only in the specific
     *         case of entry recovered from oplog ( & not rolled to Htree) & the RECOVER_VALUES flag
     *         is false . In other cases the exact value is not needed
     */
    public byte getUserBits() {
      return this.userBits;
    }

    public int getValueLength() {
      return this.valueLength;
    }

    public long getOffsetInOplog() {
      return offsetInOplog;
    }

    public long getOplogId() {
      return this.oplogId;
    }

    public void setOplogId(long v) {
      this.oplogId = v;
    }

    public boolean getValueRecovered() {
      return this.valueRecovered;
    }

    public VersionTag getVersionTag() {
      return this.tag;
    }

    public void setVersionTag(VersionTag tag) {
      this.tag = tag;
    }

    public void applyToDiskEntry(PlaceHolderDiskRegion drv, DiskEntry entry,
        RegionEntryContext context) {
      DiskId did = entry.getDiskId();
      synchronized (did) {
        applyToDiskEntry(entry, context, drv, did);
      }
    }

    public void applyToDiskEntry(DiskEntry entry, RegionEntryContext region, AbstractDiskRegion dr,
        DiskId did) {
      int oldValueLength;
      // Now that oplog creates are immediately put in cache
      // a later oplog modify will get us here
      Object oldValueAsToken = entry.getValueAsToken();
      boolean oldValueWasNull = oldValueAsToken == null;
      long oldOplogId = did.getOplogId();
      long newOplogId = getOplogId();
      if (newOplogId != oldOplogId) {
        did.setOplogId(newOplogId);
        setOplogId(oldOplogId); // so caller knows oldoplog id
      }
      did.setOffsetInOplog(getOffsetInOplog());
      // id already set
      did.setUserBits(getUserBits());
      oldValueLength = did.getValueLength();
      did.setValueLength(getValueLength());

      if (!getValueRecovered()) {
        if (!oldValueWasNull) {
          entry.handleValueOverflow(region);
          entry.setValueWithContext(region, null); // fixes bug 41119
        }
      } else {
        entry.setValueWithContext(region, entry.prepareValueForCache(region, getValue(), false));
      }

      if (!getValueRecovered()) { // recovering an entry whose new value is on disk
        if (!oldValueWasNull) { // the entry's old value is in vm
          // TODO: oldKeyId == 0 is the ILLEGAL id; what does that indicate?
          int inVM = -1;
          if (Token.isInvalidOrRemoved(oldValueAsToken)) { // but tokens are never in vm
            inVM = 0;
          }
          Helper.updateStats(dr, region, inVM, 1/* OnDisk */, did.getValueLength());
        } else { // the entry's old value is also on disk
          int valueLenDelta = -oldValueLength; // but it is no longer
          valueLenDelta += did.getValueLength(); // new one is now on disk
          Helper.updateStats(dr, region, 0, 0, valueLenDelta);
        }
      } else { // recovering an entry whose new value is in vm
        int inVM = 1;
        if (Token.isInvalidOrRemoved(getValue())) { // but tokens never in vm
          inVM = 0;
        }
        if (oldValueWasNull) { // the entry's old value is on disk
          Helper.updateStats(dr, region, inVM, -1/* OnDisk */, -oldValueLength);
        } else { // the entry's old value was in the vm
          if (inVM == 1 && Token.isInvalidOrRemoved(oldValueAsToken)) {
            // the old state was not in vm and not on disk. But now we are in vm.
            Helper.updateStats(dr, region, 1, 0, 0);
          } else if (inVM == 0 && !Token.isInvalidOrRemoved(oldValueAsToken)) {
            // the old state was in vm and not on disk. But now we are not in vm.
            Helper.updateStats(dr, region, -1, 0, 0);
          }
        }
      }
    }
  }
}