/*
 * 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.FilenameFilter;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import org.apache.logging.log4j.Logger;

import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.internal.cache.DiskStoreImpl.OplogEntryIdSet;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.internal.cache.entries.DiskEntry.Helper.ValueWrapper;
import org.apache.geode.internal.cache.persistence.DiskRecoveryStore;
import org.apache.geode.internal.cache.persistence.DiskRegionView;
import org.apache.geode.internal.cache.persistence.DiskStoreFilter;
import org.apache.geode.internal.cache.persistence.OplogType;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.sequencelog.EntryLogger;
import org.apache.geode.logging.internal.log4j.api.LogService;

public class PersistentOplogSet implements OplogSet {
  private static final Logger logger = LogService.getLogger();

  /** variable to generate sequential unique oplogEntryId's* */
  private final AtomicLong oplogEntryId = new AtomicLong(DiskStoreImpl.INVALID_ID);

  /**
   * Contains all the oplogs that only have a drf (i.e. the crf has been deleted).
   */
  private final Map<Long, Oplog> drfOnlyOplogs = new LinkedHashMap<>();

  private final Map<Long, Oplog> oplogIdToOplog = new LinkedHashMap<>();

  /** oplogs that are done being written to but not yet ready to compact */
  private final Map<Long, Oplog> inactiveOplogs = new LinkedHashMap<>(16, 0.75f, true);

  private final DiskStoreImpl parent;

  private final AtomicInteger inactiveOpenCount = new AtomicInteger();

  private final Map<Long, DiskRecoveryStore> pendingRecoveryMap = new HashMap<>();
  private final Map<Long, DiskRecoveryStore> currentRecoveryMap = new HashMap<>();

  private final AtomicBoolean alreadyRecoveredOnce = new AtomicBoolean(false);

  private final PrintStream out;

  /** The active oplog * */
  private volatile Oplog child;

  /**
   * The maximum oplog id we saw while recovering
   */
  private volatile long maxRecoveredOplogId;

  /** counter used for round-robin logic * */
  private int dirCounter = -1;

  public PersistentOplogSet(DiskStoreImpl parent, PrintStream out) {
    this.parent = parent;
    this.out = out;
  }

  /**
   * returns the active child
   */
  public Oplog getChild() {
    return child;
  }

  /**
   * set the child to a new oplog
   */
  void setChild(Oplog oplog) {
    child = oplog;
  }

  Oplog[] getAllOplogs() {
    synchronized (getOplogIdToOplog()) {
      int rollNum = getOplogIdToOplog().size();
      int inactiveNum = inactiveOplogs.size();
      int drfOnlyNum = drfOnlyOplogs.size();
      int num = rollNum + inactiveNum + drfOnlyNum + 1;
      Oplog[] oplogs = new Oplog[num];
      oplogs[0] = getChild();

      Iterator<Oplog> oplogIterator = getOplogIdToOplog().values().iterator();
      for (int i = 1; i <= rollNum; i++) {
        oplogs[i] = oplogIterator.next();
      }

      oplogIterator = inactiveOplogs.values().iterator();
      for (int i = 1; i <= inactiveNum; i++) {
        oplogs[i + rollNum] = oplogIterator.next();
      }

      oplogIterator = drfOnlyOplogs.values().iterator();
      for (int i = 1; i <= drfOnlyNum; i++) {
        oplogs[i + rollNum + inactiveNum] = oplogIterator.next();
      }

      // Special case - no oplogs found
      if (oplogs.length == 1 && oplogs[0] == null) {
        return new Oplog[0];
      }

      return oplogs;
    }

  }

  private TreeSet<Oplog> getSortedOplogs() {
    TreeSet<Oplog> result = new TreeSet<Oplog>(
        (Comparator) (arg0, arg1) -> Long
            .signum(((Oplog) arg1).getOplogId() - ((Oplog) arg0).getOplogId()));
    for (Oplog oplog : getAllOplogs()) {
      if (oplog != null) {
        result.add(oplog);
      }
    }
    return result;
  }

  /**
   * Get the oplog specified
   *
   * @param id int oplogId to be got
   * @return Oplogs the oplog corresponding to the oplodId, id
   */
  @Override
  public Oplog getChild(long id) {
    Oplog localOplog = child;
    if (localOplog != null && id == localOplog.getOplogId()) {
      return localOplog;
    }

    synchronized (getOplogIdToOplog()) {
      Oplog result = getOplogIdToOplog().get(id);
      if (result == null) {
        result = inactiveOplogs.get(id);
      }
      return result;
    }
  }

  @Override
  public void create(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async) {
    getChild().create(region, entry, value, async);
  }

  @Override
  public void modify(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async) {
    getChild().modify(region, entry, value, async);
  }

  void offlineModify(DiskRegionView drv, DiskEntry entry, byte[] value,
      boolean isSerializedObject) {
    getChild().offlineModify(drv, entry, value, isSerializedObject);
  }

  @Override
  public void remove(InternalRegion region, DiskEntry entry, boolean async, boolean isClear) {
    getChild().remove(region, entry, async, isClear);
  }

  public void forceRoll(DiskRegion dr) {
    Oplog child = getChild();
    if (child != null) {
      child.forceRolling(dr);
    }
  }

  Map<File, DirectoryHolder> findFiles(String partialFileName) {
    dirCounter = 0;
    Map<File, DirectoryHolder> backupFiles = new HashMap<>();
    FilenameFilter backupFileFilter = getFileNameFilter(partialFileName);

    for (DirectoryHolder dh : parent.directories) {
      File[] backupList = dh.getDir().listFiles(backupFileFilter);
      if (backupList != null) {
        for (File f : backupList) {
          backupFiles.put(f, dh);
        }
      }
    }

    return backupFiles;
  }

  private FilenameFilter getFileNameFilter(String partialFileName) {
    return new DiskStoreFilter(OplogType.BACKUP, false, partialFileName);
  }

  void createOplogs(boolean needsOplogs, Map<File, DirectoryHolder> backupFiles) {
    LongOpenHashSet foundCrfs = new LongOpenHashSet();
    LongOpenHashSet foundDrfs = new LongOpenHashSet();

    for (Map.Entry<File, DirectoryHolder> entry : backupFiles.entrySet()) {
      File file = entry.getKey();
      String absolutePath = file.getAbsolutePath();

      int underscorePosition = absolutePath.lastIndexOf('_');
      int pointPosition = absolutePath.lastIndexOf('.');
      String oplogIdString = absolutePath.substring(underscorePosition + 1, pointPosition);
      long oplogId = Long.parseLong(oplogIdString);

      maxRecoveredOplogId = Math.max(maxRecoveredOplogId, oplogId);

      // here look diskinit file and check if this opid already deleted or not
      // if deleted then don't process it.

      if (Oplog.isCRFFile(file.getName())) {
        if (!isCrfOplogIdPresent(oplogId)) {
          deleteFileOnRecovery(file);
          try {
            String krfFileName = Oplog.getKRFFilenameFromCRFFilename(file.getAbsolutePath());
            File krfFile = new File(krfFileName);
            deleteFileOnRecovery(krfFile);
          } catch (Exception ignore) {
            // ignore
          }
          // this file we unable to delete earlier
          continue;
        }

      } else if (Oplog.isDRFFile(file.getName())) {
        if (!isDrfOplogIdPresent(oplogId)) {
          deleteFileOnRecovery(file);
          // this file we unable to delete earlier
          continue;
        }
      }

      Oplog oplog = getChild(oplogId);
      if (oplog == null) {
        oplog = new Oplog(oplogId, this);
        addRecoveredOplog(oplog);
      }

      if (oplog.addRecoveredFile(file, entry.getValue())) {
        foundCrfs.add(oplogId);
      } else {
        foundDrfs.add(oplogId);
      }
    }
    if (needsOplogs) {
      verifyOplogs(foundCrfs, foundDrfs);
    }
  }

  private boolean isDrfOplogIdPresent(long oplogId) {
    return parent.getDiskInitFile().isDRFOplogIdPresent(oplogId);
  }

  private boolean isCrfOplogIdPresent(long oplogId) {
    return parent.getDiskInitFile().isCRFOplogIdPresent(oplogId);
  }

  private void verifyOplogs(LongOpenHashSet foundCrfs, LongOpenHashSet foundDrfs) {
    parent.getDiskInitFile().verifyOplogs(foundCrfs, foundDrfs);
  }


  private void deleteFileOnRecovery(File f) {
    try {
      f.delete();
    } catch (Exception ignore) {
      // ignore, one more attempt to delete the file failed
    }
  }

  private void addRecoveredOplog(Oplog oplog) {
    basicAddToBeCompacted(oplog);
    // don't schedule a compaction here. Wait for recovery to complete
  }

  /**
   * Taking a lock on the LinkedHashMap oplogIdToOplog as it the operation of adding an Oplog to the
   * Map & notifying the Compactor thread , if not already compaction has to be an atomic operation.
   * add the oplog to the to be compacted set. if compactor thread is active and recovery is not
   * going on then the compactor thread is notified of the addition
   */
  void addToBeCompacted(Oplog oplog) {
    basicAddToBeCompacted(oplog);
    parent.scheduleCompaction();
  }

  private void basicAddToBeCompacted(Oplog oplog) {
    if (!oplog.isRecovering() && oplog.hasNoLiveValues()) {
      oplog.cancelKrf();
      oplog.close();
      oplog.deleteFiles(oplog.getHasDeletes());

    } else {
      parent.getStats().incCompactableOplogs(1);
      Long key = oplog.getOplogId();
      int inactivePromotedCount = 0;

      synchronized (getOplogIdToOplog()) {
        if (inactiveOplogs.remove(key) != null) {
          if (oplog.isRAFOpen()) {
            inactiveOpenCount.decrementAndGet();
          }
          inactivePromotedCount++;
        }
        getOplogIdToOplog().put(key, oplog);
      }

      if (inactivePromotedCount > 0) {
        parent.getStats().incInactiveOplogs(-inactivePromotedCount);
      }
    }
  }

  void recoverRegionsThatAreReady() {
    // The following sync also prevents concurrent recoveries by multiple regions
    // which is needed currently.
    synchronized (getAlreadyRecoveredOnce()) {

      // need to take a snapshot of DiskRecoveryStores we will recover
      synchronized (pendingRecoveryMap) {
        currentRecoveryMap.clear();
        currentRecoveryMap.putAll(pendingRecoveryMap);
        pendingRecoveryMap.clear();
      }

      if (currentRecoveryMap.isEmpty() && getAlreadyRecoveredOnce().get()) {
        // no recovery needed
        return;
      }

      for (DiskRecoveryStore drs : currentRecoveryMap.values()) {
        // Call prepare early to fix bug 41119.
        drs.getDiskRegionView().prepareForRecovery();
      }

      if (!getAlreadyRecoveredOnce().get()) {
        initOplogEntryId();
        // Fix for #43026 - make sure we don't reuse an entry
        // id that has been marked as cleared.
        updateOplogEntryId(parent.getDiskInitFile().getMaxRecoveredClearEntryId());
      }

      long start = parent.getStats().startRecovery();
      EntryLogger.setSource(parent.getDiskStoreID(), "recovery");

      long byteCount = 0;
      try {
        byteCount = recoverOplogs(byteCount);

      } finally {
        Map<String, Integer> prSizes = null;
        Map<String, Integer> prBuckets = null;
        if (parent.isValidating()) {
          prSizes = new HashMap<>();
          prBuckets = new HashMap<>();
        }

        for (DiskRecoveryStore drs : currentRecoveryMap.values()) {
          for (Oplog oplog : getAllOplogs()) {
            if (oplog != null) {
              // Need to do this AFTER recovery to protect from concurrent compactions
              // trying to remove the oplogs.
              // We can't remove a dr from the oplog's unrecoveredRegionCount
              // until it is fully recovered.
              // This fixes bug 41119.
              oplog.checkForRecoverableRegion(drs.getDiskRegionView());
            }
          }

          if (parent.isValidating()) {
            if (drs instanceof ValidatingDiskRegion) {
              ValidatingDiskRegion vdr = (ValidatingDiskRegion) drs;
              if (vdr.isBucket()) {
                String prName = vdr.getPrName();
                if (prSizes.containsKey(prName)) {
                  int oldSize = prSizes.get(prName);
                  oldSize += vdr.size();
                  prSizes.put(prName, oldSize);
                  int oldBuckets = prBuckets.get(prName);
                  oldBuckets++;
                  prBuckets.put(prName, oldBuckets);
                } else {
                  prSizes.put(prName, vdr.size());
                  prBuckets.put(prName, 1);
                }
              } else {
                parent.incLiveEntryCount(vdr.size());
                out.println(vdr.getName() + ": entryCount=" + vdr.size());
              }
            }
          }
        }

        if (parent.isValidating()) {
          for (Map.Entry<String, Integer> me : prSizes.entrySet()) {
            parent.incLiveEntryCount(me.getValue());
            out.println(me.getKey() + " entryCount=" + me.getValue() + " bucketCount="
                + prBuckets.get(me.getKey()));
          }
        }

        parent.getStats().endRecovery(start, byteCount);
        getAlreadyRecoveredOnce().set(true);
        currentRecoveryMap.clear();
        EntryLogger.clearSource();
      }
    }
  }

  private long recoverOplogs(long byteCount) {
    OplogEntryIdSet deletedIds = new OplogEntryIdSet();
    TreeSet<Oplog> oplogSet = getSortedOplogs();

    if (!getAlreadyRecoveredOnce().get()) {
      if (getChild() != null && !getChild().hasBeenUsed()) {
        // Then remove the current child since it is empty
        // and does not need to be recovered from
        // and it is important to not call initAfterRecovery on it.
        oplogSet.remove(getChild());
      }
    }

    Set<Oplog> oplogsNeedingValueRecovery = new HashSet<>();

    if (!oplogSet.isEmpty()) {
      long startOpLogRecovery = System.currentTimeMillis();

      // first figure out all entries that have been destroyed
      boolean latestOplog = true;
      for (Oplog oplog : oplogSet) {
        byteCount += oplog.recoverDrf(deletedIds, getAlreadyRecoveredOnce().get(), latestOplog);
        latestOplog = false;
        if (!getAlreadyRecoveredOnce().get()) {
          updateOplogEntryId(oplog.getMaxRecoveredOplogEntryId());
        }
      }

      parent.incDeadRecordCount(deletedIds.size());

      // now figure out live entries
      latestOplog = true;
      for (Oplog oplog : oplogSet) {
        long startOpLogRead = parent.getStats().startOplogRead();
        long bytesRead = oplog.recoverCrf(deletedIds, recoverValues(), recoverValuesSync(),
            getAlreadyRecoveredOnce().get(), oplogsNeedingValueRecovery, latestOplog);
        latestOplog = false;
        if (!getAlreadyRecoveredOnce().get()) {
          updateOplogEntryId(oplog.getMaxRecoveredOplogEntryId());
        }
        byteCount += bytesRead;
        parent.getStats().endOplogRead(startOpLogRead, bytesRead);

        // Callback to the disk regions to indicate the oplog is recovered
        // Used for offline export
        for (DiskRecoveryStore drs : currentRecoveryMap.values()) {
          drs.getDiskRegionView().oplogRecovered(oplog.oplogId);
        }
      }

      long endOpLogRecovery = System.currentTimeMillis();
      long elapsed = endOpLogRecovery - startOpLogRecovery;
      logger.info("recovery oplog load took {} ms", elapsed);
    }

    if (!parent.isOfflineCompacting()) {
      long startRegionInit = System.currentTimeMillis();

      // create the oplogs now so that loadRegionData can have them available
      // Create an array of Oplogs so that we are able to add it in a single shot
      // to the map
      for (DiskRecoveryStore drs : currentRecoveryMap.values()) {
        drs.getDiskRegionView().initRecoveredEntryCount();
      }

      if (!getAlreadyRecoveredOnce().get()) {
        for (Oplog oplog : oplogSet) {
          if (oplog != getChild()) {
            oplog.initAfterRecovery(parent.isOffline());
          }
        }
        if (getChild() == null) {
          setFirstChild(getSortedOplogs(), false);
        }
      }

      if (!parent.isOffline()) {
        if (recoverValues() && !recoverValuesSync()) {
          // value recovery defers compaction because it is using the compactor thread
          parent.scheduleValueRecovery(oplogsNeedingValueRecovery, currentRecoveryMap);
        }

        if (!getAlreadyRecoveredOnce().get()) {
          // Create krfs for oplogs that are missing them
          for (Oplog oplog : oplogSet) {
            if (oplog.needsKrf()) {
              oplog.createKrfAsync();
            }
          }
          parent.scheduleCompaction();
        }

        long endRegionInit = System.currentTimeMillis();
        logger.info("recovery region initialization took {} ms", endRegionInit - startRegionInit);
      }
    }
    return byteCount;
  }

  private boolean recoverValuesSync() {
    return parent.RECOVER_VALUES_SYNC;
  }

  private boolean recoverValues() {
    return parent.RECOVER_VALUES;
  }

  private void setFirstChild(TreeSet<Oplog> oplogSet, boolean force) {
    if (parent.isOffline() && !parent.isOfflineCompacting() && !parent.isOfflineModify()) {
      return;
    }

    if (!oplogSet.isEmpty()) {
      Oplog first = oplogSet.first();
      DirectoryHolder dh = first.getDirectoryHolder();
      dirCounter = dh.getArrayIndex();
      dirCounter = ++dirCounter % parent.dirLength;
      // we want the first child to go in the directory after the directory
      // used by the existing oplog with the max id.
      // This fixes bug 41822.
    }

    if (force || maxRecoveredOplogId > 0) {
      setChild(new Oplog(maxRecoveredOplogId + 1, this, getNextDir()));
    }
  }

  private void initOplogEntryId() {
    oplogEntryId.set(DiskStoreImpl.INVALID_ID);
  }

  /**
   * Sets the last created oplogEntryId to the given value if and only if the given value is greater
   * than the current last created oplogEntryId
   */
  private void updateOplogEntryId(long v) {
    long curVal;
    do {
      curVal = oplogEntryId.get();
      if (curVal >= v) {
        // no need to set
        return;
      }
    } while (!oplogEntryId.compareAndSet(curVal, v));
  }

  /**
   * Returns the last created oplogEntryId. Returns INVALID_ID if no oplogEntryId has been created.
   */
  long getOplogEntryId() {
    parent.initializeIfNeeded();
    return oplogEntryId.get();
  }

  /**
   * Creates and returns a new oplogEntryId for the given key. An oplogEntryId is needed when
   * storing a key/value pair on disk. A new one is only needed if the key is new. Otherwise the
   * oplogEntryId already allocated for a key can be reused for the same key.
   *
   * @return A disk id that can be used to access this key/value pair on disk
   */
  long newOplogEntryId() {
    return oplogEntryId.incrementAndGet();
  }

  /**
   * Returns the next available DirectoryHolder which has space. If no dir has space then it will
   * return one anyway if compaction is enabled.
   *
   * @param minAvailableSpace the minimum amount of space we need in this directory.
   */
  DirectoryHolder getNextDir(long minAvailableSpace, boolean checkForWarning) {
    if (minAvailableSpace < parent.getMaxOplogSizeInBytes()
        && !DiskStoreImpl.SET_IGNORE_PREALLOCATE) {
      minAvailableSpace = parent.getMaxOplogSizeInBytes();
    }

    DirectoryHolder selectedHolder = null;
    synchronized (parent.getDirectoryHolders()) {
      for (int i = 0; i < parent.dirLength; ++i) {
        DirectoryHolder dirHolder = parent.directories[dirCounter];

        // Increment the directory counter to next position so that next time when this operation
        // is invoked, it checks for the Directory Space in a cyclical fashion.
        dirCounter = ++dirCounter % parent.dirLength;

        // if the current directory has some space, then quit and return the dir
        if (dirHolder.getAvailableSpace() >= minAvailableSpace) {
          if (checkForWarning && !parent.isDirectoryUsageNormal(dirHolder)) {
            if (logger.isDebugEnabled()) {
              logger.debug("Ignoring directory {} due to insufficient disk space", dirHolder);
            }
          } else {
            selectedHolder = dirHolder;
            break;
          }
        }
      }

      if (selectedHolder == null) {
        // we didn't find a warning-free directory, try again ignoring the check
        if (checkForWarning) {
          return getNextDir(minAvailableSpace, false);
        }

        if (parent.isCompactionEnabled()) {
          selectedHolder = parent.directories[dirCounter];

          // Increment the directory counter to next position
          dirCounter = ++dirCounter % parent.dirLength;
          if (selectedHolder.getAvailableSpace() < minAvailableSpace) {
            logger.warn(
                "Even though the configured directory size limit has been exceeded a new oplog will be created because compaction is enabled. The configured limit is {}. The current space used in the directory by this disk store is {}.",
                selectedHolder.getUsedSpace(), selectedHolder.getCapacity());
          }
        } else {
          throw new DiskAccessException(
              "Disk is full and compaction is disabled. No space can be created", parent);
        }
      }
    }

    return selectedHolder;
  }

  DirectoryHolder getNextDir() {
    return getNextDir(DiskStoreImpl.MINIMUM_DIR_SIZE, true);
  }

  private void addDrf(Oplog oplog) {
    synchronized (getOplogIdToOplog()) {
      drfOnlyOplogs.put(oplog.getOplogId(), oplog);
    }
  }

  void removeDrf(Oplog oplog) {
    synchronized (getOplogIdToOplog()) {
      drfOnlyOplogs.remove(oplog.getOplogId());
    }
  }

  /**
   * Return true if id is less than all the ids in the oplogIdToOplog map. Since the oldest one is
   * in the LINKED hash map is first we only need to compare ourselves to it.
   */
  boolean isOldestExistingOplog(long id) {
    synchronized (getOplogIdToOplog()) {
      for (long otherId : getOplogIdToOplog().keySet()) {
        if (id > otherId) {
          return false;
        }
      }

      // since the inactiveOplogs map is an LRU we need to check each one
      for (long otherId : inactiveOplogs.keySet()) {
        if (id > otherId) {
          return false;
        }
      }
    }

    return true;
  }

  /**
   * Destroy all the oplogs that are:
   *
   * <pre>
   * 1. the oldest (based on smallest oplog id)
   * 2. empty (have no live values)
   * </pre>
   */
  void destroyOldestReadyToCompact() {
    synchronized (getOplogIdToOplog()) {
      if (drfOnlyOplogs.isEmpty()) {
        return;
      }
    }

    Oplog oldestLiveOplog = getOldestLiveOplog();
    List<Oplog> toDestroy = new ArrayList<>();

    if (oldestLiveOplog == null) {
      // remove all oplogs that are empty
      synchronized (getOplogIdToOplog()) {
        toDestroy.addAll(drfOnlyOplogs.values());
      }

    } else {
      // remove all empty oplogs that are older than the oldest live one
      synchronized (getOplogIdToOplog()) {
        for (Oplog oplog : drfOnlyOplogs.values()) {
          if (oplog.getOplogId() < oldestLiveOplog.getOplogId()) {
            toDestroy.add(oplog);
          }
        }
      }
    }

    for (Oplog oplog : toDestroy) {
      oplog.destroy();
    }
  }

  private Oplog getOldestLiveOplog() {
    Oplog result = null;
    synchronized (getOplogIdToOplog()) {
      for (Oplog oplog : getOplogIdToOplog().values()) {
        if (result == null || oplog.getOplogId() < result.getOplogId()) {
          result = oplog;
        }
      }

      // since the inactiveOplogs map is an LRU we need to check each one
      for (Oplog oplog : inactiveOplogs.values()) {
        if (result == null || oplog.getOplogId() < result.getOplogId()) {
          result = oplog;
        }
      }
    }

    return result;
  }

  void inactiveAccessed(Oplog oplog) {
    Long key = oplog.getOplogId();
    synchronized (getOplogIdToOplog()) {
      // update last access time
      inactiveOplogs.get(key);
    }
  }

  void inactiveReopened(Oplog oplog) {
    addInactive(oplog, true);
  }

  void addInactive(Oplog oplog) {
    addInactive(oplog, false);
  }

  private void addInactive(Oplog oplog, boolean reopen) {
    Long key = oplog.getOplogId();
    List<Oplog> openlist = null;
    synchronized (getOplogIdToOplog()) {

      boolean isInactive = true;
      if (reopen) {
        // It is possible that 'oplog' is compactible instead of inactive. So set isInactive.
        isInactive = inactiveOplogs.get(key) != null;
      } else {
        inactiveOplogs.put(key, oplog);
      }

      if (reopen && isInactive || oplog.isRAFOpen()) {
        if (inactiveOpenCount.incrementAndGet() > DiskStoreImpl.MAX_OPEN_INACTIVE_OPLOGS) {
          openlist = new ArrayList<>();
          for (Oplog inactiveOplog : inactiveOplogs.values()) {
            if (inactiveOplog.isRAFOpen()) {
              // add to my list
              openlist.add(inactiveOplog);
            }
          }
        }
      }
    }

    if (openlist != null) {
      for (Oplog openOplog : openlist) {
        if (openOplog.closeRAF()) {
          if (inactiveOpenCount.decrementAndGet() <= DiskStoreImpl.MAX_OPEN_INACTIVE_OPLOGS) {
            break;
          }
        }
      }
    }

    if (!reopen) {
      parent.getStats().incInactiveOplogs(1);
    }
  }

  public void clear(DiskRegion diskRegion, RegionVersionVector regionVersionVector) {
    // call clear on each oplog
    Collection<Oplog> oplogsToClear = new ArrayList<>();
    synchronized (getOplogIdToOplog()) {
      oplogsToClear.addAll(getOplogIdToOplog().values());
      oplogsToClear.addAll(inactiveOplogs.values());

      Oplog child = getChild();
      if (child != null) {
        oplogsToClear.add(child);
      }
    }

    for (Oplog oplog : oplogsToClear) {
      oplog.clear(diskRegion, regionVersionVector);
    }

    if (regionVersionVector != null) {
      parent.getDiskInitFile().clearRegion(diskRegion, regionVersionVector);
    } else {
      long clearedOplogEntryId = getOplogEntryId();
      parent.getDiskInitFile().clearRegion(diskRegion, clearedOplogEntryId);
    }
  }

  public RuntimeException close() {
    RuntimeException firstRuntimeException = null;
    try {
      closeOtherOplogs();
    } catch (RuntimeException e) {
      firstRuntimeException = e;
    }

    if (child != null) {
      try {
        child.finishKrf();
      } catch (RuntimeException e) {
        if (firstRuntimeException != null) {
          firstRuntimeException = e;
        }
      }

      try {
        child.close();
      } catch (RuntimeException e) {
        if (firstRuntimeException != null) {
          firstRuntimeException = e;
        }
      }
    }

    return firstRuntimeException;
  }

  /** closes all the oplogs except the current one * */
  private void closeOtherOplogs() {
    // get a snapshot to prevent CME
    Oplog[] oplogs = getAllOplogs();

    // if there are oplogs which are to be compacted, destroy them do not do oplogs[0]
    for (int i = 1; i < oplogs.length; i++) {
      oplogs[i].finishKrf();
      oplogs[i].close();
      removeOplog(oplogs[i].getOplogId());
    }
  }

  /**
   * Removes the oplog from the map given the oplogId
   *
   * @param id id of the oplog to be removed from the list
   * @return oplog Oplog which has been removed
   */
  Oplog removeOplog(long id) {
    return removeOplog(id, false, null);
  }

  Oplog removeOplog(long id, boolean deleting, Oplog olgToAddToDrfOnly) {
    Oplog oplog;
    boolean drfOnly = false;
    boolean inactive = false;

    synchronized (getOplogIdToOplog()) {
      Long key = id;
      oplog = getOplogIdToOplog().remove(key);
      if (oplog == null) {
        oplog = inactiveOplogs.remove(key);
        if (oplog != null) {
          if (oplog.isRAFOpen()) {
            inactiveOpenCount.decrementAndGet();
          }
          inactive = true;
        } else {
          oplog = drfOnlyOplogs.remove(key);
          if (oplog != null) {
            drfOnly = true;
          }
        }
      }

      if (olgToAddToDrfOnly != null) {
        addDrf(olgToAddToDrfOnly);
      }
    }

    if (oplog != null) {
      if (!drfOnly) {
        if (inactive) {
          parent.getStats().incInactiveOplogs(-1);
        } else {
          parent.getStats().incCompactableOplogs(-1);
        }
      }

      if (!deleting && !oplog.isOplogEmpty()) {
        // we are removing an oplog whose files are not deleted
        parent.undeletedOplogSize.addAndGet(oplog.getOplogSize());
      }
    }
    return oplog;
  }

  public void basicClose(DiskRegion dr) {
    List<Oplog> oplogsToClose = new ArrayList<>();
    synchronized (getOplogIdToOplog()) {
      oplogsToClose.addAll(getOplogIdToOplog().values());
      oplogsToClose.addAll(inactiveOplogs.values());
      oplogsToClose.addAll(drfOnlyOplogs.values());

      Oplog child = getChild();
      if (child != null) {
        oplogsToClose.add(child);
      }
    }

    for (Oplog oplog : oplogsToClose) {
      oplog.close(dr);
    }
  }

  public void prepareForClose() {
    Collection<Oplog> oplogsToPrepare = new ArrayList<>();
    synchronized (getOplogIdToOplog()) {
      oplogsToPrepare.addAll(getOplogIdToOplog().values());
      oplogsToPrepare.addAll(inactiveOplogs.values());
    }

    boolean childPreparedForClose = false;
    long childOplogId = getChild() == null ? -1 : getChild().oplogId;

    for (Oplog oplog : oplogsToPrepare) {
      oplog.prepareForClose();
      if (childOplogId != -1 && oplog.oplogId == childOplogId) {
        childPreparedForClose = true;
      }
    }

    if (!childPreparedForClose && getChild() != null) {
      getChild().prepareForClose();
    }
  }

  public void basicDestroy(DiskRegion diskRegion) {
    Collection<Oplog> oplogsToDestroy = new ArrayList<>();
    synchronized (getOplogIdToOplog()) {
      oplogsToDestroy.addAll(getOplogIdToOplog().values());
      oplogsToDestroy.addAll(inactiveOplogs.values());
      oplogsToDestroy.addAll(drfOnlyOplogs.values());

      Oplog child = getChild();
      if (child != null) {
        oplogsToDestroy.add(child);
      }
    }

    for (Oplog oplog : oplogsToDestroy) {
      oplog.destroy(diskRegion);
    }
  }

  void destroyAllOplogs() {
    // get a snapshot to prevent ConcurrentModificationException
    for (Oplog oplog : getAllOplogs()) {
      if (oplog != null) {
        oplog.destroy();
        removeOplog(oplog.getOplogId());
      }
    }
  }

  /**
   * Add compactable oplogs to the list, up to the maximum size.
   */
  void getCompactableOplogs(List<CompactableOplog> compactableOplogs, int max) {
    synchronized (getOplogIdToOplog()) {
      for (Oplog oplog : getOplogIdToOplog().values()) {
        if (compactableOplogs.size() >= max) {
          return;
        }
        if (oplog.needsCompaction()) {
          compactableOplogs.add(oplog);
        }
      }
    }
  }

  void scheduleForRecovery(DiskRecoveryStore diskRecoveryStore) {
    DiskRegionView diskRegionView = diskRecoveryStore.getDiskRegionView();
    if (diskRegionView.isRecreated() &&
        (diskRegionView.getMyPersistentID() != null
            || diskRegionView.getMyInitializingID() != null)) {
      // If a region does not have either id then don't pay the cost
      // of scanning the oplogs for recovered data.
      synchronized (pendingRecoveryMap) {
        pendingRecoveryMap.put(diskRegionView.getId(), diskRecoveryStore);
      }
    }
  }

  /**
   * Returns null if we are not currently recovering the DiskRegion with the given drId.
   */
  DiskRecoveryStore getCurrentlyRecovering(long drId) {
    return currentRecoveryMap.get(drId);
  }

  void initChild() {
    if (getChild() == null) {
      setFirstChild(getSortedOplogs(), true);
    }
  }

  public void offlineCompact() {
    if (getChild() != null) {
      // check active oplog and if it is empty delete it
      getChild().krfClose();
      if (getChild().isOplogEmpty()) {
        getChild().destroy();
      }
    }

    // remove any oplogs that only have a drf
    Collection<Oplog> oplogsToDestroy = new ArrayList<>();
    synchronized (getOplogIdToOplog()) {
      for (Oplog oplog : getOplogIdToOplog().values()) {
        if (oplog.isDrfOnly()) {
          oplogsToDestroy.add(oplog);
        }
      }
    }

    for (Oplog oplog : oplogsToDestroy) {
      oplog.destroy();
    }

    destroyOldestReadyToCompact();
  }

  public DiskStoreImpl getParent() {
    return parent;
  }

  void updateDiskRegion(AbstractDiskRegion diskRegion) {
    for (Oplog oplog : getAllOplogs()) {
      if (oplog != null) {
        oplog.updateDiskRegion(diskRegion);
      }
    }
  }

  void flushChild() {
    Oplog oplog = getChild();
    if (oplog != null) {
      oplog.flushAll();
    }
  }

  public String getPrefix() {
    return OplogType.BACKUP.getPrefix();
  }

  void crfCreate(long oplogId) {
    getParent().getDiskInitFile().crfCreate(oplogId);
  }

  void drfCreate(long oplogId) {
    getParent().getDiskInitFile().drfCreate(oplogId);
  }

  void crfDelete(long oplogId) {
    getParent().getDiskInitFile().crfDelete(oplogId);
  }

  void drfDelete(long oplogId) {
    getParent().getDiskInitFile().drfDelete(oplogId);
  }

  boolean couldHaveKrf() {
    return getParent().couldHaveKrf();
  }

  public boolean isCompactionPossible() {
    return getParent().isCompactionPossible();
  }

  /** oplogs that are ready to compact */
  Map<Long, Oplog> getOplogIdToOplog() {
    return oplogIdToOplog;
  }

  AtomicBoolean getAlreadyRecoveredOnce() {
    return alreadyRecoveredOnce;
  }

  @VisibleForTesting
  Map<Long, DiskRecoveryStore> getPendingRecoveryMap() {
    return pendingRecoveryMap;
  }
}