/*
 * 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.internal.Assert;
import org.apache.geode.internal.HeapDataOutputStream;
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.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.serialization.ByteArrayDataInput;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.util.BlobHelper;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * 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
     */
    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) {
      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);
            } 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,
        RegionEntryContext context,
        KnownVersion version) {
      @Retained
      @Released
      Object value;
      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(de.getLastModified());

          ReferenceCountHelper.setReferenceCountOwner(entry);

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

          ReferenceCountHelper.setReferenceCountOwner(null);
          if (value == null) {
            if (did == null) {
              synchronized (de) {
                did = de.getDiskId();
              }
              assert did != null;
              // do recursive call to get readLock on did
              return fillInValue(de, entry, dr, context, version);
            }
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "DiskEntry.Helper.fillInValue, key={}; getting value from disk, disk id={}",
                  entry.getKey(), did);
            }
            final BytesAndBits bb;
            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(value)) {
        return false;
      } else if (value instanceof CachedDeserializable) {
        CachedDeserializable cd = (CachedDeserializable) value;
        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(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 (value != entry.getValue()) {
            OffHeapHelper.releaseWithNoTracking(value);
          }
        }
      } else if (value instanceof byte[]) {
        entry.setValue(value);
        entry.setSerialized(false);
      } else if (value == Token.INVALID) {
        entry.setInvalid();
      } else if (value == Token.LOCAL_INVALID) {
        entry.setLocalInvalid();
      } else if (value == Token.TOMBSTONE) {
        entry.setTombstone();
      } else {
        Object preparedValue = value;
        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) {
            throw new IllegalArgumentException("An IOException was thrown while serializing.", e);
          }
        }
      }
      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 isSerializedObject;
      }

      @Override
      public int getLength() {
        return (bytes != null) ? 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 (bytes == null) {
              throw new IllegalStateException("userBits==1 and value is null");
            } else if (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(bytes, offset, bytesThisTime);
          offset += bytesThisTime;
          if (needsFlush) {
            flushable.flush();
          }
        }
      }

      @Override
      public String getBytesAsString() {
        if (bytes == null) {
          return "null";
        }
        StringBuilder sb = new StringBuilder();
        int len = getLength();
        for (int i = 0; i < len; i++) {
          sb.append(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 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();
        offHeapData = so;
      }

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

      @Override
      public int getLength() {
        return 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 = 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 = 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 = 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 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) {
        return ((StoredObject) o).hasRefCount();
      }
      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
        final Object 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
      final 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 {

        DiskRegion dr = region.getDiskRegion();
        DiskId did = entry.getDiskId();
        Object syncObj = did;
        if (syncObj == null) {
          syncObj = entry;
        }
        if (syncObj == did) {
          dr.acquireReadLock();
        }

        final AsyncDiskEntry asyncDiskEntry;
        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);
        }
        boolean caughtCacheClosed = false;
        try {
          if (!Token.isRemovedFromDisk(oldValue)) {
            result = basicRemoveFromDisk(entry, region, false);
          }
        } catch (CacheClosedException e) {
          caughtCacheClosed = true;
          throw e;
        } finally {
          // do not to set the value to be removedFromDisk since it failed to persist
          if (!caughtCacheClosed) {
            // 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.
          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);
