/*
 * 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;

import java.io.File;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.logging.log4j.Logger;

import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.distributed.OplogCancelledException;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.DiskStoreImpl.OplogCompactor;
import org.apache.geode.internal.cache.Oplog.OplogDiskEntry;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.internal.cache.entries.DiskEntry.Helper.Flushable;
import org.apache.geode.internal.cache.entries.DiskEntry.Helper.ValueWrapper;
import org.apache.geode.internal.cache.persistence.BytesAndBits;
import org.apache.geode.internal.cache.persistence.DiskRegionView;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * An oplog used for overflow-only regions. For regions that are persistent (i.e. they can be
 * recovered) see {@link Oplog}.
 *
 * @since GemFire prPersistSprint2
 */
class OverflowOplog implements CompactableOplog, Flushable {
  private static final Logger logger = LogService.getLogger();

  /* System property to override the disk store write buffer size. */
  public final String WRITE_BUFFER_SIZE_SYS_PROP_NAME = "WRITE_BUF_SIZE";

  /** Extension of the oplog file * */
  static final String CRF_FILE_EXT = ".crf";

  /** The file which will be created on disk * */
  private final File diskFile;

  /** boolean marked true when this oplog is closed * */
  private volatile boolean closed;

  private final OplogFile crf = new OplogFile();

  private final ByteBuffer[] bbArray = new ByteBuffer[2];

  /** The stats for this store */
  private final DiskStoreStats stats;

  /** The store that owns this Oplog* */
  private final DiskStoreImpl parent;

  /**
   * The oplog set this oplog is part of
   */
  private final OverflowOplogSet oplogSet;

  /** oplog id * */
  protected final int oplogId;

  /** Directory in which the file is present* */
  private final DirectoryHolder dirHolder;

  /**
   * The max Oplog size (user configurable). Set to zero when oplog is deleted.
   */
  private long maxOplogSize;

  private final OpState opState;

  /**
   * Set to true when this oplog will no longer be written to. Never set to false once it becomes
   * true.
   */
  private boolean doneAppending = false;

  private final OplogDiskEntry liveEntries = new OplogDiskEntry();

  /**
   * A copy constructor used for creating a new oplog based on the previous Oplog. This constructor
   * is invoked only from the function switchOplog
   *
   * @param oplogId integer identifying the new oplog
   * @param dirHolder The directory in which to create new Oplog
   * @param minSize Minimum oplog file size in bytes
   */
  OverflowOplog(int oplogId, OverflowOplogSet parent, DirectoryHolder dirHolder, long minSize) {
    this.oplogId = oplogId;
    this.parent = parent.getParent();
    oplogSet = parent;
    this.dirHolder = dirHolder;
    opState = new OpState();
    long maxOplogSizeParam = this.parent.getMaxOplogSizeInBytes();
    if (maxOplogSizeParam < minSize) {
      maxOplogSizeParam = minSize;
    }
    long availableSpace = this.dirHolder.getAvailableSpace();
    if (availableSpace < minSize
        // fix for bug 42464
        && !this.parent.isCompactionEnabled()) {
      availableSpace = minSize;
    }
    if (availableSpace < maxOplogSizeParam
        // fix for bug 42464
        && !this.parent.isCompactionEnabled() && availableSpace > 0) {
      // createOverflowOplog decided that we should use this dir. So do it.
      maxOplogSize = availableSpace;
    } else {
      maxOplogSize = maxOplogSizeParam;
    }
    stats = this.parent.getStats();

    closed = false;
    String n = this.parent.getName();
    diskFile = new File(this.dirHolder.getDir(), "OVERFLOW" + n + "_" + oplogId);
    try {
      createCrf(parent.getActiveOverflowOplog());
    } catch (IOException ex) {
      throw new DiskAccessException(
          String.format("Failed creating operation log because: %s", ex),
          this.parent);
    }
  }

  DiskStoreImpl getParent() {
    return parent;
  }

  private OverflowOplogSet getOplogSet() {
    return oplogSet;
  }

  private void preblow() {
    dirHolder.incrementTotalOplogSize(maxOplogSize);
    final OplogFile olf = getOLF();
    try {
      olf.raf.setLength(maxOplogSize);
      olf.raf.seek(0);
    } catch (IOException ignore) {
      // TODO: need a warning since this can impact perf.
      // I don't think I need any of this. If setLength throws then
      // the file is still ok.
    }
  }

  /**
   * Creates the crf oplog file
   */
  private void createCrf(OverflowOplog previous) throws IOException {
    File f = new File(diskFile.getPath() + CRF_FILE_EXT);

    if (logger.isDebugEnabled()) {
      logger.debug("Creating operation log file {}", f);
    }
    crf.f = f;
    crf.raf = new RandomAccessFile(f, "rw");
    crf.writeBuf = allocateWriteBuf(previous);
    preblow();
    logger.info("Created {} {} for disk store {}.",
        new Object[] {toString(), "crf", parent.getName()});
    crf.channel = crf.raf.getChannel();

    stats.incOpenOplogs();
  }

  @VisibleForTesting
  Integer getWriteBufferSizeProperty() {
    return Integer.getInteger(WRITE_BUFFER_SIZE_SYS_PROP_NAME);
  }

  @VisibleForTesting
  Integer getWriteBufferCapacity() {
    Integer writeBufferSizeProperty = getWriteBufferSizeProperty();
    if (writeBufferSizeProperty != null) {
      return writeBufferSizeProperty;
    }
    return getParent().getWriteBufferSize();
  }

  private ByteBuffer allocateWriteBuf(OverflowOplog previous) {
    ByteBuffer result = null;
    if (previous != null) {
      result = previous.consumeWriteBuf();
    }
    if (result == null) {
      return ByteBuffer.allocateDirect(getWriteBufferCapacity());
    }
    return result;
  }

  private ByteBuffer consumeWriteBuf() {
    synchronized (crf) {
      ByteBuffer result = crf.writeBuf;
      crf.writeBuf = null;
      return result;
    }
  }

  /**
   * Returns the <code>DiskStoreStats</code> for this oplog
   */
  public DiskStoreStats getStats() {
    return stats;
  }

  /**
   * Test Method to be used only for testing purposes. Gets the underlying File object for the Oplog
   * . Oplog class uses this File object to obtain the RandomAccessFile object. Before returning the
   * File object , the dat present in the buffers of the RandomAccessFile object is flushed.
   * Otherwise, for windows the actual file length does not match with the File size obtained from
   * the File object
   */
  File getOplogFile() throws IOException {
    synchronized (crf) {
      if (!crf.RAFClosed) {
        crf.raf.getFD().sync();
      }
      return crf.f;
    }
  }

  /** the oplog identifier * */
  public int getOplogId() {
    return oplogId;
  }

  /**
   * Returns the unserialized bytes and bits for the given Entry. If Oplog is destroyed while
   * querying, then the DiskRegion is queried again to obatin the value This method should never get
   * invoked for an entry which has been destroyed
   *
   * @since GemFire 3.2.1
   * @param id The DiskId for the entry @param offset The offset in this OpLog where the entry is
   *        present. @param faultingIn @param bitOnly boolean indicating whether to extract just the
   *        UserBit or UserBit with value @return BytesAndBits object wrapping the value & user bit
   */
  @Override
  public BytesAndBits getBytesAndBits(DiskRegionView dr, DiskId id, boolean faultingIn,
      boolean bitOnly) {
    OverflowOplog retryOplog = null;
    long offset = 0;
    synchronized (id) {
      int opId = (int) id.getOplogId();
      if (opId != getOplogId()) {
        // the oplog changed on us so we need to do a recursive
        // call after unsyncing
        retryOplog = getOplogSet().getChild(opId);
      } else {
        // fetch this while synced so it will be consistent with oplogId
        offset = id.getOffsetInOplog();
      }
    }
    if (retryOplog != null) {
      return retryOplog.getBytesAndBits(dr, id, faultingIn, bitOnly);
    }
    BytesAndBits bb = null;
    long start = stats.startRead();

    // If the offset happens to be -1, still it is possible that
    // the data is present in the current oplog file.
    if (offset == -1) {
      // Since it is given that a get operation has alreadty
      // taken a
      // lock on an entry , no put operation could have modified the
      // oplog ID
      // there fore synchronization is not needed
      // synchronized (id) {
      offset = id.getOffsetInOplog();
    }

    // If the current OpLog is not destroyed ( its opLogRaf file
    // is still open) we can retrieve the value from this oplog.
    try {
      bb = basicGet(dr, offset, bitOnly, id.getValueLength(), id.getUserBits());
    } catch (DiskAccessException dae) {
      logger.error(
          String.format("Oplog::basicGet: Error in reading the data from disk for Disk ID %s",
              id),
          dae);
      throw dae;
    }
    // If bb is still null then entry has been compacted to the Htree
    // or in case of concurrent get & put , to a new OpLog ( Concurrent Get
    // &
    // Put is not possible at this point).
    // Since the compacter takes a lock on Entry as well as DiskId , the
    // situation below
    // will not be possible and hence commenting the code

    if (bb == null) {
      throw new EntryDestroyedException(
          String.format(
              "No value was found for entry with disk Id %s on a region  with synchronous writing set to %s",
              id, dr.isSync()));
    }
    if (bitOnly) {
      dr.endRead(start, stats.endRead(start, 1), 1);
    } else {
      dr.endRead(start, stats.endRead(start, bb.getBytes().length), bb.getBytes().length);
    }
    return bb;
  }

  /**
   * Returns the object stored on disk with the given id. This method is used for testing purposes
   * only. As such, it bypasses the buffer and goes directly to the disk. This is not a thread safe
   * function , in the sense, it is possible that by the time the OpLog is queried , data might move
   * HTree with the oplog being destroyed
   *
   * @param id A DiskId object for which the value on disk will be fetched
   */
  @Override
  public BytesAndBits getNoBuffer(DiskRegion dr, DiskId id) {
    if (logger.isTraceEnabled()) {
      logger.trace("Oplog::getNoBuffer:Before invoking Oplog.basicGet for DiskID ={}", id);
    }

    try {
      return basicGet(dr, id.getOffsetInOplog(), false, id.getValueLength(), id.getUserBits());
    } catch (DiskAccessException dae) {
      logger.error(
          String.format("Oplog::getNoBuffer:Exception in retrieving value from disk for diskId=%s",
              id),
          dae);
      throw dae;
    } catch (IllegalStateException ise) {
      logger.error(
          String.format("Oplog::getNoBuffer:Exception in retrieving value from disk for diskId=%s",
              id),
          ise);
      throw ise;
    }
  }

  void freeEntry(DiskEntry de) {
    rmLive(de);
  }

  /**
   * Call this when the cache is closed or region is destroyed. Deletes the lock files and if it is
   * Overflow only, deletes the oplog file as well
   */
  public void close() {
    if (closed) {
      return;
    }
    if (logger.isDebugEnabled()) {
      logger.debug("Oplog::close: Store name ={} Oplog ID = {}", parent.getName(), oplogId);
    }

    basicClose();
  }

  /**
   * Close the files of a oplog but don't set any state. Used by unit tests
   */
  public void testClose() {
    try {
      crf.channel.close();
    } catch (IOException ignore) {
    }
    try {
      crf.raf.close();
    } catch (IOException ignore) {
    }
  }

  private void basicClose() {
    flushAll();
    synchronized (crf) {
      if (!crf.RAFClosed) {
        try {
          crf.channel.close();
        } catch (IOException ignore) {
        }
        try {
          crf.raf.close();
        } catch (IOException ignore) {
        }
        crf.RAFClosed = true;
        stats.decOpenOplogs();
      }
      closed = true;
    }

    deleteFiles();
  }

  /**
   * Destroys this oplog. First it will call close which will cleanly close all Async threads. The
   * deletion of lock files will be taken care of by the close. Close will also take care of
   * deleting the files if it is overflow only mode
   *
   */
  public void destroy() {
    if (!closed) {
      lockCompactor();
      try {
        basicClose();
      } finally {
        unlockCompactor();
      }
    }
  }

  /**
   * A check to confirm that the oplog has been closed because of the cache being closed
   *
   */
  private void checkClosed() {
    parent.getCancelCriterion().checkCancelInProgress(null);
    if (!closed) {
      return;
    }
    throw new OplogCancelledException("This Oplog has been closed.");
  }

  /**
   * Return absolute value of v.
   */
  static long abs(long v) {
    if (v < 0) {
      return -v;
    } else {
      return v;
    }
  }

  private void initOpState(DiskEntry entry, ValueWrapper value, byte userBits) {
    opState.initialize(entry, value, userBits);
  }

  private void clearOpState() {
    opState.clear();
  }

  /**
   * Returns the number of bytes it will take to serialize this.opState.
   */
  private int getOpStateSize() {
    return opState.getSize();
  }

  private byte calcUserBits(ValueWrapper value) {
    return value.getUserBits();
  }

  /**
   * Modifies a key/value pair from a region entry on disk. Updates all of the necessary
   * {@linkplain DiskStoreStats statistics} and invokes basicModify
   * <p>
   * Modified the code so as to reuse the already created ByteBuffer during transition. Minimizing
   * the synchronization allowing multiple put operations for different entries to proceed
   * concurrently for asynch mode
   *
   * @param entry DiskEntry object representing the current Entry
   * @param value byte array representing the value
   * @return true if modify was done; false if this file did not have room
   */
  public boolean modify(DiskRegion dr, DiskEntry entry, ValueWrapper value, boolean async) {
    try {
      byte userBits = calcUserBits(value);
      return basicModify(entry, value, userBits, async);
    } catch (IOException ex) {
      throw new DiskAccessException(
          String.format("Failed writing key to %s", diskFile.getPath()),
          ex, dr.getName());
    } catch (InterruptedException ie) {
      Thread.currentThread().interrupt();
      dr.getCancelCriterion().checkCancelInProgress(ie);
      throw new DiskAccessException(
          String.format(
              "Failed writing key to %s due to failure in acquiring read lock for asynch writing",
              diskFile.getPath()),
          ie, dr.getName());
    }
  }

  public boolean copyForwardForOverflowCompact(DiskEntry entry, byte[] value, int length,
      byte userBits) {
    try {
      ValueWrapper vw = new DiskEntry.Helper.CompactorValueWrapper(value, length);
      return basicModify(entry, vw, userBits, true);
    } catch (IOException ex) {
      throw new DiskAccessException(
          String.format("Failed writing key to %s", diskFile.getPath()),
          ex, getParent().getName());
    } catch (InterruptedException ie) {
      Thread.currentThread().interrupt();
      getParent().getCancelCriterion().checkCancelInProgress(ie);
      throw new DiskAccessException(
          String.format(
              "Failed writing key to %s due to failure in acquiring read lock for asynch writing",
              diskFile.getPath()),
          ie, getParent().getName());
    }
  }


  /**
   * A helper function which identifies whether to modify the entry in the current oplog or to make
   * the switch to the next oplog. This function enables us to reuse the byte buffer which got
   * created for an oplog which no longer permits us to use itself. It will also take acre of
   * compaction if required
   *
   * @param entry DiskEntry object representing the current Entry
   * @return true if modify was done; false if this file did not have room
   */
  private boolean basicModify(DiskEntry entry, ValueWrapper value, byte userBits, boolean async)
      throws IOException, InterruptedException {
    DiskId id = entry.getDiskId();
    long startPosForSynchOp = -1L;
    OverflowOplog emptyOplog = null;
    synchronized (crf) {
      initOpState(entry, value, userBits);
      int adjustment = getOpStateSize();
      assert adjustment > 0;
      int oldOplogId;
      // do the io while holding lock so that switch can set doneAppending
      // Write the data to the opLog for the synch mode
      startPosForSynchOp = writeOpLogBytes(async);
      if (startPosForSynchOp == -1) {
        return false;
      } else {
        if (logger.isTraceEnabled()) {
          logger.trace("Oplog::basicModify:Released ByteBuffer with data for Disk ID = {}", id);
        }
        synchronized (id) {
          // Need to do this while synced on id
          oldOplogId = (int) id.setOplogId(getOplogId());
          id.setOffsetInOplog(startPosForSynchOp);
          if (EntryBits.isNeedsValue(userBits)) {
            id.setValueLength(value.getLength());
          } else {
            id.setValueLength(0);
          }
          id.setUserBits(userBits);
        }
        // Note: we always call rmLive (and addLive) even if this mod was to an entry
        // last modified in this same oplog to cause the list to be sorted by offset
        // so when the compactor iterates over it will read values with a sequential scan
        // instead of hopping around the oplog.
        if (oldOplogId > 0) {
          OverflowOplog oldOplog = getOplogSet().getChild(oldOplogId);
          if (oldOplog != null) {
            if (oldOplog.rmLive(entry)) {
              if (oldOplogId != getOplogId()) {
                emptyOplog = oldOplog;
              }
            }
          }
        }
        addLive(entry);
      }
      clearOpState();
    }
    if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
      CacheObserverHolder.getInstance().afterSettingOplogOffSet(startPosForSynchOp);
    }
    if (emptyOplog != null
        && (!emptyOplog.isCompacting() || emptyOplog.calledByCompactorThread())) {
      if (emptyOplog.calledByCompactorThread() && emptyOplog.hasNoLiveValues()) {
        flushAll();
      }
      emptyOplog.handleNoLiveValues();
    }
    return true;
  }

  /**
   * Removes the key/value pair with the given id on disk.
   *
   * @param entry DiskEntry object on which remove operation is called
   */
  public void remove(DiskRegion dr, DiskEntry entry) {
    try {
      basicRemove(dr, entry);
    } catch (IOException ex) {
      throw new DiskAccessException(
          String.format("Failed writing key to %s", diskFile.getPath()),
          ex, dr.getName());
    } catch (InterruptedException ie) {
      Thread.currentThread().interrupt();
      dr.getCancelCriterion().checkCancelInProgress(ie);
      throw new DiskAccessException(
          String.format(
              "Failed writing key to %s due to failure in acquiring read lock for asynch writing",
              diskFile.getPath()),
          ie, dr.getName());
    }
  }

  /**
   * A helper function which identifies whether to record a removal of entry in the current oplog or
   * to make the switch to the next oplog. This function enables us to reuse the byte buffer which
   * got created for an oplog which no longer permits us to use itself.
   *
   * @param entry DiskEntry object representing the current Entry
   */
  private void basicRemove(DiskRegion dr, DiskEntry entry)
      throws IOException, InterruptedException {
    DiskId id = entry.getDiskId();

    // no need to lock since this code no longer does io
    if (EntryBits.isNeedsValue(id.getUserBits())) {
      // oplogId already done up in DiskStoreImpl
      long oldOffset = id.getOffsetInOplog();
      if (oldOffset != -1) {
        id.setOffsetInOplog(-1);
        if (rmLive(entry)) {
          if (!isCompacting() || calledByCompactorThread()) {
            handleNoLiveValues();
          }
        }
      }
    }
  }

  /**
   * test hook
   */
  public ByteBuffer getWriteBuf() {
    return crf.writeBuf;
  }

  private static final int MAX_CHANNEL_RETRIES = 5;

  @Override
  public void flush() throws IOException {
    final OplogFile olf = crf;
    synchronized (olf) {
      if (olf.RAFClosed) {
        return;
      }
      try {
        ByteBuffer bb = olf.writeBuf;
        if (bb != null && bb.position() != 0) {
          bb.flip();
          int flushed = 0;
          int numChannelRetries = 0;
          do {
            int channelBytesWritten = 0;
            final int bbStartPos = bb.position();
            final long channelStartPos = olf.channel.position();
            // differentiate between bytes written on this channel.write() iteration and the
            // total number of bytes written to the channel on this call
            channelBytesWritten = olf.channel.write(bb);
            // Expect channelBytesWritten and the changes in pp.position() and channel.position() to
            // be the same. If they are not, then the channel.write() silently failed. The following
            // retry separates spurious failures from permanent channel failures.
            if (channelBytesWritten != bb.position() - bbStartPos) {
              if (numChannelRetries++ < MAX_CHANNEL_RETRIES) {
                // Reset the ByteBuffer position, but take into account anything that did get
                // written to the channel
                channelBytesWritten = (int) (olf.channel.position() - channelStartPos);
                bb.position(bbStartPos + channelBytesWritten);
              } else {
                throw new IOException("Failed to write Oplog entry to " + olf.f.getName() + ": "
                    + "channel.write() returned " + channelBytesWritten + ", "
                    + "change in channel position = " + (olf.channel.position() - channelStartPos)
                    + ", " + "change in source buffer position = " + (bb.position() - bbStartPos));
              }
            }
            flushed += channelBytesWritten;
          } while (bb.hasRemaining());
          // update bytesFlushed after entire writeBuffer is flushed to fix bug 41201
          olf.bytesFlushed += flushed;
          bb.clear();
        }
      } catch (ClosedChannelException ignore) {
        // It is possible for a channel to be closed when our code does not
        // explicitly call channel.close (when we will set RAFclosed).
        // This can happen when a thread is doing an io op and is interrupted.
        // That thread will see ClosedByInterruptException but it will also
        // close the channel and then we will see ClosedChannelException.
      }
    }
  }

  /**
   * Method to be used only for testing
   *
   * @param ch Object to replace the channel in the Oplog.crf
   * @return original channel object
   */
  FileChannel testSetCrfChannel(FileChannel ch) {
    FileChannel chPrev = crf.channel;
    crf.channel = ch;
    return chPrev;
  }

  @Override
  public void flush(ByteBuffer b1, ByteBuffer b2) throws IOException {
    final OplogFile olf = crf;
    synchronized (olf) {
      if (olf.RAFClosed) {
        return;
      }
      try {
        bbArray[0] = b1;
        bbArray[1] = b2;
        b1.flip();
        long flushed = 0;
        do {
          flushed += olf.channel.write(bbArray);
        } while (b2.hasRemaining());
        bbArray[0] = null;
        bbArray[1] = null;
        // update bytesFlushed after entire writeBuffer is flushed to fix bug 41201
        olf.bytesFlushed += flushed;
        b1.clear();
      } catch (ClosedChannelException ignore) {
        // It is possible for a channel to be closed when our code does not
        // explicitly call channel.close (when we will set RAFclosed).
        // This can happen when a thread is doing an io op and is interrupted.
        // That thread will see ClosedByInterruptException but it will also
        // close the channel and then we will see ClosedChannelException.
      }
    }
  }

  public void flushAll() {
    try {
      flush();
    } catch (IOException ex) {
      throw new DiskAccessException(
          String.format("Failed writing key to %s", diskFile.getPath()),
          ex, parent);
    }
  }

  /**
   * Since the ByteBuffer being writen to can have additional bytes which are used for extending the
   * size of the file, it is necessary that the ByteBuffer provided should have limit which is set
   * to the position till which it contains the actual bytes. If the mode is synched write then only
   * we will write up to the capacity & opLogSpace variable have any meaning. For asynch mode it
   * will be zero. Also this method must be synchronized on the file , whether we use synch or
   * asynch write because the fault in operations can clash with the asynch writing. Write the
   * specified bytes to the oplog. Note that since extending a file is expensive this code will
   * possibly write OPLOG_EXTEND_SIZE zero bytes to reduce the number of times the file is extended.
   *
   *
   * @return The long offset at which the data present in the ByteBuffer gets written to
   */
  private long writeOpLogBytes(boolean async) throws IOException {
    long startPos = -1L;
    final OplogFile olf = crf;
    synchronized (olf) {
      if (doneAppending) {
        return -1;
      }
      if (closed) {
        Assert.assertTrue(false, this + " for store " + parent.getName()
            + " has been closed for synch mode while writing is going on. This should not happen");
      }
      // It is assumed that the file pointer is already at the
      // appropriate position in the file so as to allow writing at the end.
      // Any fault in operations will set the pointer back to the write location.
      // Also it is only in case of synch writing, we are writing more
      // than what is actually needed, we will have to reset the pointer.
      // Also need to add in offset in writeBuf in case we are not flushing writeBuf
      long curFileOffset = olf.channel.position() + olf.writeBuf.position();
      startPos = allocate(curFileOffset, getOpStateSize());
      if (startPos != -1) {
        if (startPos != curFileOffset) {
          flush();
          olf.channel.position(startPos);
          olf.bytesFlushed = startPos;
          stats.incOplogSeeks();
        }
        if (logger.isTraceEnabled(LogMarker.PERSIST_WRITES_VERBOSE)) {
          logger.trace(LogMarker.PERSIST_WRITES_VERBOSE, "writeOpLogBytes startPos={} oplog#{}",
              startPos, getOplogId());
        }
        long oldBytesFlushed = olf.bytesFlushed;
        long bytesWritten = opState.write();
        if ((startPos + bytesWritten) > olf.currSize) {
          olf.currSize = startPos + bytesWritten;
        }
        if (logger.isTraceEnabled(LogMarker.PERSIST_WRITES_VERBOSE)) {
          logger.trace(LogMarker.PERSIST_WRITES_VERBOSE,
              "writeOpLogBytes bytesWritten={} oldBytesFlushed={} byteFlushed={} oplog#{}",
              bytesWritten, oldBytesFlushed, olf.bytesFlushed, getOplogId());
        }
        if (oldBytesFlushed != olf.bytesFlushed) {
          // opState.write must have done an implicit flush
          // so we need to do an explicit flush so the value
          // can be read back in entirely from disk.
          flush();
        }
        getStats().incWrittenBytes(bytesWritten, async);

        // // Moved the set of lastWritePos to after write
        // // so if write throws an exception it will not be updated.
        // // This fixes bug 40449.
      }
    }
    return startPos;
  }

  private BytesAndBits attemptGet(DiskRegionView dr, long offsetInOplog, int valueLength,
      byte userBits) throws IOException {
    synchronized (crf) {
      assert offsetInOplog >= 0;
      RandomAccessFile myRAF = crf.raf;
      BytesAndBits bb = null;
      long writePosition = 0;
      if (!doneAppending) {
        writePosition = myRAF.getFilePointer();
        bb = attemptWriteBufferGet(writePosition, offsetInOplog, valueLength, userBits);
        if (bb == null) {
          if ((offsetInOplog + valueLength) > crf.bytesFlushed && !closed) {
            flushAll(); // fix for bug 41205
            writePosition = myRAF.getFilePointer();
          }
        }
      }
      if (bb == null) {
        myRAF.seek(offsetInOplog);
        try {
          stats.incOplogSeeks();
          byte[] valueBytes = new byte[valueLength];
          myRAF.readFully(valueBytes);
          stats.incOplogReads();
          bb = new BytesAndBits(valueBytes, userBits);
        } finally {
          // if this oplog is no longer being appended to then don't waste disk io
          if (!doneAppending) {
            myRAF.seek(writePosition);
            stats.incOplogSeeks();
          }
        }
      }
      return bb;
    } // sync
  }

  private BytesAndBits attemptWriteBufferGet(long writePosition, long readPosition, int valueLength,
      byte userBits) {
    BytesAndBits bb = null;
    ByteBuffer writeBuf = crf.writeBuf;
    int curWriteBufPos = writeBuf.position();
    if (writePosition <= readPosition
        && (writePosition + curWriteBufPos) >= (readPosition + valueLength)) {
      int bufOffset = (int) (readPosition - writePosition);
      byte[] valueBytes = new byte[valueLength];
      int oldLimit = writeBuf.limit();
      writeBuf.limit(curWriteBufPos);
      writeBuf.position(bufOffset);
      writeBuf.get(valueBytes);
      writeBuf.position(curWriteBufPos);
      writeBuf.limit(oldLimit);
      bb = new BytesAndBits(valueBytes, userBits);
    }
    return bb;
  }

  /**
   * Extracts the Value byte array & UserBit from the OpLog
   *
   * @param offsetInOplog The starting position from which to read the data in the opLog
   * @param bitOnly boolean indicating whether the value needs to be extracted along with the
   *        UserBit or not.
   * @param valueLength The length of the byte array which represents the value
   * @param userBits The userBits of the value.
   * @return BytesAndBits object which wraps the extracted value & user bit
   */
  private BytesAndBits basicGet(DiskRegionView dr, long offsetInOplog, boolean bitOnly,
      int valueLength, byte userBits) {
    BytesAndBits bb = null;
    if (EntryBits.isAnyInvalid(userBits) || EntryBits.isTombstone(userBits) || bitOnly
        || valueLength == 0) {
      if (EntryBits.isInvalid(userBits)) {
        bb = new BytesAndBits(DiskEntry.INVALID_BYTES, userBits);
      } else if (EntryBits.isTombstone(userBits)) {
        bb = new BytesAndBits(DiskEntry.TOMBSTONE_BYTES, userBits);
      } else {
        bb = new BytesAndBits(DiskEntry.LOCAL_INVALID_BYTES, userBits);
      }
    } else {
      if (offsetInOplog == -1) {
        return null;
      }
      try {
        for (;;) {
          dr.getCancelCriterion().checkCancelInProgress(null);
          boolean interrupted = Thread.interrupted();
          try {
            bb = attemptGet(dr, offsetInOplog, valueLength, userBits);
            break;
          } catch (InterruptedIOException ignore) {
            // bug 39756
            // ignore, we'll clear and retry.
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
          }
