/*
 * 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 static org.apache.geode.cache.Region.SEPARATOR_CHAR;
import static org.apache.geode.distributed.ConfigurationProperties.CACHE_XML_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.internal.cache.entries.DiskEntry.Helper.readRawValue;
import static org.apache.geode.internal.monitoring.ThreadsMonitoring.Mode.AsyncWriterExecutor;

import java.io.File;
import java.io.FileOutputStream;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.PrintStream;
import java.math.BigDecimal;
import java.net.InetAddress;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import org.apache.commons.io.FileUtils;
import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.TestOnly;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.persistence.PersistentID;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.backup.BackupService;
import org.apache.geode.internal.cache.backup.DiskStoreBackup;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.internal.cache.entries.DiskEntry.Helper.ValueWrapper;
import org.apache.geode.internal.cache.entries.DiskEntry.RecoveredEntry;
import org.apache.geode.internal.cache.eviction.AbstractEvictionController;
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.persistence.DiskStoreFilter;
import org.apache.geode.internal.cache.persistence.DiskStoreID;
import org.apache.geode.internal.cache.persistence.OplogType;
import org.apache.geode.internal.cache.persistence.PRPersistentConfig;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.persistence.PersistentMemberPattern;
import org.apache.geode.internal.cache.snapshot.GFSnapshot;
import org.apache.geode.internal.cache.snapshot.GFSnapshot.SnapshotWriter;
import org.apache.geode.internal.cache.snapshot.SnapshotPacket.SnapshotRecord;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
import org.apache.geode.internal.monitoring.executor.AbstractExecutor;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.util.BlobHelper;
import org.apache.geode.logging.internal.executors.LoggingExecutors;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.internal.ManagementConstants;
import org.apache.geode.pdx.internal.EnumInfo;
import org.apache.geode.pdx.internal.PdxField;
import org.apache.geode.pdx.internal.PdxType;
import org.apache.geode.pdx.internal.PeerTypeRegistration;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Represents a (disk-based) persistent store for region data. Used for both persistent recoverable
 * regions and overflow-only regions.
 *
 * @since GemFire 3.2
 */
@SuppressWarnings("synthetic-access")
public class DiskStoreImpl implements DiskStore {
  private static final Logger logger = LogService.getLogger();

  public static final boolean KRF_DEBUG = Boolean.getBoolean("disk.KRF_DEBUG");

  public static final int MAX_OPEN_INACTIVE_OPLOGS =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "MAX_OPEN_INACTIVE_OPLOGS", 7);

  /*
   * If less than 20MB (default - configurable through this property) of the available space is left
   * for logging and other misc stuff then it is better to bail out.
   */
  public static final int MIN_DISK_SPACE_FOR_LOGS =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "MIN_DISK_SPACE_FOR_LOGS", 20);

  /** Represents an invalid id of a key/value on disk */
  public static final long INVALID_ID = 0L; // must be zero

  public static final String COMPLETE_COMPACTION_BEFORE_TERMINATION_PROPERTY_NAME =
      GeodeGlossary.GEMFIRE_PREFIX + "disk.completeCompactionBeforeTermination";

  static final int MINIMUM_DIR_SIZE = 1024;

  private DiskDirSizesUnit diskDirSizesUnit;

  /**
   * The static field delays the joining of the close/clear/destroy & forceFlush operation, with the
   * compactor thread. This joining occurs after the compactor thread is notified to exit. This was
   * added to reproduce deadlock caused by concurrent destroy & clear operation where clear
   * operation is restarting the compactor thread ( a new thread object different from the one for
   * which destroy operation issued notification for release). The delay occurs iff the flag used
   * for enabling callbacks to CacheObserver is enabled true
   */
  @MutableForTesting
  static volatile long DEBUG_DELAY_JOINING_WITH_COMPACTOR = 500;

  /**
   * Kept for backwards compat. Should use allowForceCompaction api/dtd instead.
   */
  private static final boolean ENABLE_NOTIFY_TO_ROLL =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "ENABLE_NOTIFY_TO_ROLL");

  public static final String RECOVER_VALUE_PROPERTY_NAME =
      GeodeGlossary.GEMFIRE_PREFIX + "disk.recoverValues";

  public static final String RECOVER_VALUES_SYNC_PROPERTY_NAME =
      GeodeGlossary.GEMFIRE_PREFIX + "disk.recoverValuesSync";

  /**
   * When configured threshold value is reached, then server will overflow to
   * the new hashmap during the recovery of .drf files
   */
  public static final String DRF_HASHMAP_OVERFLOW_THRESHOLD_NAME =
      GeodeGlossary.GEMFIRE_PREFIX + "disk.drfHashMapOverflowThreshold";

  /**
   * Allows recovering values for LRU regions. By default values are not recovered for LRU regions
   * during recovery.
   */
  public static final String RECOVER_LRU_VALUES_PROPERTY_NAME =
      GeodeGlossary.GEMFIRE_PREFIX + "disk.recoverLruValues";

  static final long DRF_HASHMAP_OVERFLOW_THRESHOLD_DEFAULT = 805306368;
  final long DRF_HASHMAP_OVERFLOW_THRESHOLD =
      Long.getLong(DRF_HASHMAP_OVERFLOW_THRESHOLD_NAME, DRF_HASHMAP_OVERFLOW_THRESHOLD_DEFAULT);

  boolean RECOVER_VALUES = getBoolean(DiskStoreImpl.RECOVER_VALUE_PROPERTY_NAME, true);

  boolean RECOVER_VALUES_SYNC = getBoolean(DiskStoreImpl.RECOVER_VALUES_SYNC_PROPERTY_NAME, false);

  boolean FORCE_KRF_RECOVERY =
      getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "disk.FORCE_KRF_RECOVERY", false);

  final boolean RECOVER_LRU_VALUES =
      getBoolean(DiskStoreImpl.RECOVER_LRU_VALUES_PROPERTY_NAME, false);

  public static boolean getBoolean(String sysProp, boolean def) {
    return Boolean.parseBoolean(System.getProperty(sysProp, Boolean.valueOf(def).toString()));
  }

  public static final long MIN_RESERVED_DRID = 1;

  public static final long MAX_RESERVED_DRID = 8;

  static final long MIN_DRID = MAX_RESERVED_DRID + 1;

  /**
   * Maximum number of oplogs to compact per compaction operations. Defaults to 1 to allows oplogs
   * to be deleted quickly, to reduce amount of memory used during a compaction and to be fair to
   * other regions waiting for a compactor thread from the pool. Ignored if set to <= 0. Made non
   * static so tests can set it.
   */
  private final int MAX_OPLOGS_PER_COMPACTION = Integer.getInteger(
      GeodeGlossary.GEMFIRE_PREFIX + "MAX_OPLOGS_PER_COMPACTION",
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "MAX_OPLOGS_PER_ROLL", 1).intValue());

  public static final int MAX_CONCURRENT_COMPACTIONS = Integer.getInteger(
      GeodeGlossary.GEMFIRE_PREFIX + "MAX_CONCURRENT_COMPACTIONS",
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "MAX_CONCURRENT_ROLLS", 1).intValue());

  /**
   * This system property indicates that maximum number of delayed write tasks that can be pending
   * before submitting the tasks start blocking. These tasks are things like unpreblow oplogs,
   * delete oplogs, etc.
   */
  public static final int MAX_PENDING_TASKS =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "disk.MAX_PENDING_TASKS", 6);

  /**
   * This system property indicates that IF should also be preallocated. This property will be used
   * in conjunction with the PREALLOCATE_OPLOGS property. If PREALLOCATE_OPLOGS is ON the below will
   * by default be ON but in order to switch it off you need to explicitly
   */
  static final boolean PREALLOCATE_IF =
      !System.getProperty(GeodeGlossary.GEMFIRE_PREFIX + "preAllocateIF", "true")
          .equalsIgnoreCase("false");

  /**
   * This system property indicates that Oplogs should be preallocated till the maxOplogSize as
   * specified for the disk store.
   */
  static final boolean PREALLOCATE_OPLOGS =
      !System.getProperty(GeodeGlossary.GEMFIRE_PREFIX + "preAllocateDisk", "true")
          .equalsIgnoreCase("false");

  /**
   * For some testing purposes we would not consider top property if this flag is set to true
   **/
  @MutableForTesting
  public static boolean SET_IGNORE_PREALLOCATE = false;

  /**
   * This system property turns on synchronous writes just the the init file.
   */
  static final boolean SYNC_IF_WRITES =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "syncMetaDataWrites");

  /**
   * For testing - to keep track of files for which fallocate happened
   */
  @MutableForTesting
  public static volatile HashSet<String> TEST_CHK_FALLOC_DIRS;
  @MutableForTesting
  public static volatile HashSet<String> TEST_NO_FALLOC_DIRS;

  private final InternalCache cache;

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

  /**
   * Added as stop gap arrangement to fix bug 39380. It is not a clean fix as keeping track of the
   * threads acquiring read lock, etc is not a good idea to solve the issue
   */
  private final AtomicInteger entryOpsCount = new AtomicInteger();
  /**
   * Do not want to take chance with any object like DiskRegion etc as lock
   */
  private final Object closeRegionGuard = new Object();

  /** Number of dirs* */
  final int dirLength;

  /** Disk directory holders* */
  DirectoryHolder[] directories;

  /** max of all the dir sizes given stored in bytes* */
  private final long maxDirSize;

  /** total sum of dir sizes in bytes* */
  private long totalDiskStoreSpace;

  /** disk dir to be used by info file * */
  private int infoFileDirIndex;

  private final int compactionThreshold;

  /**
   * The limit of how many items can be in the async queue before async starts blocking and a flush
   * is forced. If this value is 0 then no limit.
   */
  private final int maxAsyncItems;

  private final AtomicInteger forceFlushCount;

  private final Object asyncMonitor;

  /** Compactor task which does the compaction. Null if compaction not possible. */
  private final OplogCompactor oplogCompactor;

  private DiskInitFile initFile = null;

  private final ReentrantReadWriteLock compactorLock = new ReentrantReadWriteLock();

  private final WriteLock compactorWriteLock = compactorLock.writeLock();

  private final ReadLock compactorReadLock = compactorLock.readLock();

  /**
   * Set if we have encountered a disk exception causing us to shutdown this disk store. This is
   * currently used only to prevent trying to shutdown the disk store from multiple threads, but I
   * think at some point we should use this to prevent any other ops from completing during the
   * close operation.
   */
  private final AtomicReference<DiskAccessException> diskException =
      new AtomicReference<>();

  private final PersistentOplogSet persistentOplogs = new PersistentOplogSet(this, System.out);

  OverflowOplogSet overflowOplogs = new OverflowOplogSet(this);

  private final AtomicLong regionIdCtr = new AtomicLong(MIN_DRID);

  /**
   * Only contains backup DiskRegions. The Value could be a RecoveredDiskRegion or a DiskRegion
   */
  private final ConcurrentMap<Long, DiskRegion> drMap = new ConcurrentHashMap<>();

  /**
   * A set of overflow only regions that are using this disk store.
   */
  private final Set<DiskRegion> overflowMap = ConcurrentHashMap.newKeySet();

  /**
   * Contains all of the disk recovery stores for which we are recovering values asnynchronously.
   */
  private final Map<Long, DiskRecoveryStore> currentAsyncValueRecoveryMap =
      new HashMap<>();

  private final Object asyncValueRecoveryLock = new Object();

  /**
   * The unique id for this disk store.
   *
   * Either set during recovery of an existing disk store when the IFREC_DISKSTORE_ID record is read
   * or when a new init file is created.
   *
   */
  private DiskStoreID diskStoreID;

  private final CountDownLatch _testHandleDiskAccessException = new CountDownLatch(1);

  private final ExecutorService diskStoreTaskPool;

  private final ExecutorService delayedWritePool;

  private volatile Future<?> lastDelayedWrite;

  private boolean dataCorrupted;

  private static int calcCompactionThreshold(int ct) {
    if (ct == DiskStoreFactory.DEFAULT_COMPACTION_THRESHOLD) {
      // allow the old sys prop for backwards compat.
      if (System
          .getProperty(GeodeGlossary.GEMFIRE_PREFIX + "OVERFLOW_ROLL_PERCENTAGE") != null) {
        ct = (int) (Double.parseDouble(System
            .getProperty(GeodeGlossary.GEMFIRE_PREFIX + "OVERFLOW_ROLL_PERCENTAGE", "0.50"))
            * 100.0);
      }
    }
    return ct;
  }

  /**
   * Creates a new {@code DiskRegion} that access disk on behalf of the given region.
   */
  DiskStoreImpl(InternalCache cache, DiskStoreAttributes props) {
    this(cache, props, false, null);
  }

  DiskStoreImpl(InternalCache cache, DiskStoreAttributes props, boolean ownedByRegion,
      InternalRegionArguments internalRegionArgs) {
    this(cache, props.getName(), props, ownedByRegion, internalRegionArgs, false,
        false/* upgradeVersionOnly */, false, false, true, false, /* offlineModify */
        cache.getInternalDistributedSystem().getStatisticsManager(),
        cache.getInternalResourceManager());
  }

  DiskStoreImpl(InternalCache cache, String name, DiskStoreAttributes props, boolean ownedByRegion,
      InternalRegionArguments internalRegionArgs, boolean offline,
      boolean upgradeVersionOnly,
      boolean offlineValidating, boolean offlineCompacting, boolean needsOplogs,
      boolean offlineModify, StatisticsFactory statisticsFactory,
      InternalResourceManager internalResourceManager) {
    this.offline = offline;
    this.upgradeVersionOnly = upgradeVersionOnly;
    validating = offlineValidating;
    this.offlineCompacting = offlineCompacting;
    this.offlineModify = offlineModify;
    this.internalResourceManager = internalResourceManager;

    assert internalRegionArgs == null || ownedByRegion : "internalRegionArgs "
        + "should be non-null only if the DiskStore is owned by region";
    this.ownedByRegion = ownedByRegion;
    this.internalRegionArgs = internalRegionArgs;

    this.name = name;
    autoCompact = props.getAutoCompact();
    allowForceCompaction = props.getAllowForceCompaction();
    compactionThreshold = calcCompactionThreshold(props.getCompactionThreshold());
    maxOplogSizeInBytes = props.getMaxOplogSizeInBytes();
    timeInterval = props.getTimeInterval();
    queueSize = props.getQueueSize();
    writeBufferSize = props.getWriteBufferSize();
    diskDirs = props.getDiskDirs();
    diskDirSizes = props.getDiskDirSizes();
    diskDirSizesUnit = props.getDiskDirSizesUnit();
    warningPercent = props.getDiskUsageWarningPercentage();
    criticalPercent = props.getDiskUsageCriticalPercentage();

    this.cache = cache;
    stats = new DiskStoreStats(statisticsFactory, getName());

    // start simple init

    this.dataCorrupted = false;

    isCompactionPossible = isOfflineCompacting() || (!isOffline()
        && (getAutoCompact() || getAllowForceCompaction() || ENABLE_NOTIFY_TO_ROLL));
    maxAsyncItems = getQueueSize();
    forceFlushCount = new AtomicInteger();
    asyncMonitor = new Object();
    // always use LinkedBlockingQueue to work around bug 41470
    // if (this.maxAsyncItems > 0 && this.maxAsyncItems < 1000000) {
    // // we compare to 1,000,000 so that very large maxItems will
    // // not cause us to consume too much memory in our queue.
    // // Removed the +13 since it made the queue bigger than was configured.
    // // The +13 is to give us a bit of headroom during the drain.
    // this.asyncQueue = new
    // ArrayBlockingQueue<Object>(this.maxAsyncItems/*+13*/);
    // } else {
    if (maxAsyncItems > 0) {
      asyncQueue = new ForceableLinkedBlockingQueue<>(maxAsyncItems); // fix for bug
                                                                      // 41310
    } else {
      asyncQueue = new ForceableLinkedBlockingQueue<>();
    }
    if (!isOffline()) {
      startAsyncFlusher();
    }

    File[] dirs = getDiskDirs();
    int[] dirSizes = getDiskDirSizes();
    int length = dirs.length;
    directories = new DirectoryHolder[length];
    long tempMaxDirSize = 0;
    totalDiskStoreSpace = 0;

    for (int i = 0; i < length; i++) {
      directories[i] =
          new DirectoryHolder(getName() + "_DIR#" + i, statisticsFactory, dirs[i], dirSizes[i], i,
              diskDirSizesUnit);

      if (tempMaxDirSize < dirSizes[i]) {
        tempMaxDirSize = dirSizes[i];
      }
      if (dirSizes[i] == DiskStoreFactory.DEFAULT_DISK_DIR_SIZE) {
        totalDiskStoreSpace = ManagementConstants.NOT_AVAILABLE_LONG;
      } else if (totalDiskStoreSpace != ManagementConstants.NOT_AVAILABLE_LONG) {
        totalDiskStoreSpace += (1024 * 1024) * ((long) dirSizes[i]);
      }
    }
    // stored in bytes
    maxDirSize = tempMaxDirSize * 1024 * 1024;
    infoFileDirIndex = 0;
    // Now that we no longer have db files, use all directories for oplogs
    /*
     * The infoFileDir contains the lock file and the init file. It will be directories[0] on a
     * brand new disk store. On an existing disk store it will be the directory the init file is
     * found in.
     */
    dirLength = length;

    loadFiles(needsOplogs);

    // setFirstChild(getSortedOplogs());

    // complex init
    if (isCompactionPossible() && !isOfflineCompacting()) {
      oplogCompactor = new OplogCompactor();
      oplogCompactor.startCompactor();
    } else {
      oplogCompactor = null;
    }

    diskStoreTaskPool = LoggingExecutors.newFixedThreadPoolWithFeedSize(
        MAX_CONCURRENT_COMPACTIONS, Integer.MAX_VALUE, "Idle OplogCompactor");
    delayedWritePool =
        LoggingExecutors.newFixedThreadPoolWithFeedSize(1, MAX_PENDING_TASKS, "Oplog Delete Task");
  }

  // //////////////////// Instance Methods //////////////////////

  public boolean sameAs(DiskStoreAttributes props) {
    if (getAllowForceCompaction() != props.getAllowForceCompaction()) {
      if (logger.isDebugEnabled()) {
        logger.debug("allowForceCompaction {} != {}", getAllowForceCompaction(),
            props.getAllowForceCompaction());
      }
    }
    if (getAutoCompact() != props.getAutoCompact()) {
      if (logger.isDebugEnabled()) {
        logger.debug("AutoCompact {} != {}", getAutoCompact(), props.getAutoCompact());
      }
    }
    if (getCompactionThreshold() != props.getCompactionThreshold()) {
      if (logger.isDebugEnabled()) {
        logger.debug("CompactionThreshold {} != {}", getCompactionThreshold(),
            props.getCompactionThreshold());
      }
    }
    if (getMaxOplogSizeInBytes() != props.getMaxOplogSizeInBytes()) {
      if (logger.isDebugEnabled()) {
        logger.debug("MaxOplogSizeInBytes {} != {}", getMaxOplogSizeInBytes(),
            props.getMaxOplogSizeInBytes());
      }
    }
    if (!getName().equals(props.getName())) {
      if (logger.isDebugEnabled()) {
        logger.debug("Name {} != {}", getName(), props.getName());
      }
    }
    if (getQueueSize() != props.getQueueSize()) {
      if (logger.isDebugEnabled()) {
        logger.debug("QueueSize {} != {}", getQueueSize(), props.getQueueSize());
      }
    }
    if (getTimeInterval() != props.getTimeInterval()) {
      if (logger.isDebugEnabled()) {
        logger.debug("TimeInterval {} != {}", getTimeInterval(), props.getTimeInterval());
      }
    }
    if (getWriteBufferSize() != props.getWriteBufferSize()) {
      logger.debug("WriteBufferSize {} != {}", getWriteBufferSize(), props.getWriteBufferSize());
    }
    if (!Arrays.equals(getDiskDirs(), props.getDiskDirs())) {
      if (logger.isDebugEnabled()) {
        logger.debug("DiskDirs {} != {}", Arrays.toString(getDiskDirs()),
            Arrays.toString(props.getDiskDirs()));
      }
    }
    if (!Arrays.equals(getDiskDirSizes(), props.getDiskDirSizes())) {
      if (logger.isDebugEnabled()) {
        logger.debug("DiskDirSizes {} != {}", Arrays.toString(getDiskDirSizes()),
            Arrays.toString(props.getDiskDirSizes()));
      }
    }

    return getAllowForceCompaction() == props.getAllowForceCompaction()
        && getAutoCompact() == props.getAutoCompact()
        && getCompactionThreshold() == props.getCompactionThreshold()
        && getMaxOplogSizeInBytes() == props.getMaxOplogSizeInBytes()
        && getName().equals(props.getName()) && getQueueSize() == props.getQueueSize()
        && getTimeInterval() == props.getTimeInterval()
        && getWriteBufferSize() == props.getWriteBufferSize()
        && Arrays.equals(getDiskDirs(), props.getDiskDirs())
        && Arrays.equals(getDiskDirSizes(), props.getDiskDirSizes());
  }

  /**
   * Returns the {@code DiskStoreStats} for this store
   */
  public DiskStoreStats getStats() {
    return stats;
  }

  public Map<Long, AbstractDiskRegion> getAllDiskRegions() {
    Map<Long, AbstractDiskRegion> results = new HashMap<>();
    results.putAll(drMap);
    results.putAll(initFile.getDRMap());
    return results;
  }

  void scheduleForRecovery(DiskRecoveryStore drs) {
    DiskRegionView dr = drs.getDiskRegionView();
    PersistentOplogSet oplogSet = getPersistentOplogSet(dr);
    oplogSet.scheduleForRecovery(drs);
  }

  /**
   * Initializes the contents of any regions on this DiskStore that have been registered but are not
   * yet initialized.
   */
  void initializeOwner(LocalRegion lr) {
    DiskRegion dr = lr.getDiskRegion();
    // We don't need to do recovery for overflow regions.
    if (!lr.getDataPolicy().withPersistence() || !dr.isRecreated()) {
      return;
    }

    // prevent async recovery from recovering a value
    // while we are copying the entry map.
    synchronized (currentAsyncValueRecoveryMap) {
      DiskRegionView drv = lr.getDiskRegionView();
      if (drv.getRecoveredEntryMap() != null) {
        PersistentOplogSet oplogSet = getPersistentOplogSet(drv);

        // acquire CompactorWriteLock only if the region attributes for the
        // real region are different from the place holder region's
        boolean releaseCompactorWriteLock = false;
        if (drv.isEntriesMapIncompatible()) {
          acquireCompactorWriteLock(); // fix bug #51097 to prevent concurrent compaction
          releaseCompactorWriteLock = true;
        }
        try {
          drv.copyExistingRegionMap(lr);
          getStats().incUncreatedRecoveredRegions(-1);
          for (Oplog oplog : oplogSet.getAllOplogs()) {
            if (oplog != null) {
              oplog.updateDiskRegion(lr.getDiskRegionView());
            }
          }
        } finally {
          if (releaseCompactorWriteLock) {
            releaseCompactorWriteLock();
          }
        }
        if (currentAsyncValueRecoveryMap.containsKey(drv.getId())) {
          currentAsyncValueRecoveryMap.put(drv.getId(), lr);
        }
        return;
      }
    }

    scheduleForRecovery(lr);

    // boolean gotLock = false;

    try {
      // acquireReadLock(dr);
      // gotLock = true;
      recoverRegionsThatAreReady();
    } catch (DiskAccessException dae) {
      // Asif:Just rethrow t
      throw dae;
    } catch (RuntimeException re) {
      // @todo: if re is caused by a RegionDestroyedException
      // (or CacheClosed...) then don't we want to throw that instead
      // of a DiskAccessException?
      // Asif :wrap it in DiskAccessException
      // IOException is alerady wrappped by DiskRegion correctly.
      // Howvever EntryEventImpl .deserialize is converting IOException
      // into IllegalArgumentExcepption, so handle only run time exception
      // here
      throw new DiskAccessException("RuntimeException in initializing the disk store from the disk",
          re, this);
    }
    // finally {
    // if(gotLock) {
    // releaseReadLock(dr);
    // }
    // }
  }

  private OplogSet getOplogSet(DiskRegionView drv) {
    if (drv.isBackup()) {
      return getPersistentOplogs();
    } else {
      return overflowOplogs;
    }
  }

  public PersistentOplogSet getPersistentOplogSet() {
    return getPersistentOplogs();
  }

  PersistentOplogSet getPersistentOplogSet(DiskRegionView drv) {
    assert drv.isBackup();
    return getPersistentOplogs();
  }

  /**
   * Stores a key/value pair from a region entry on disk. Updates all of the necessary
   * {@linkplain DiskRegionStats statistics}and invokes {@link Oplog#create}or {@link Oplog#modify}.
   *
   * @param entry The entry which is going to be written to disk
   * @throws RegionClearedException If a clear operation completed before the put operation
   *         completed successfully, resulting in the put operation to abort.
   * @throws IllegalArgumentException If {@code id} is less than zero
   */
  void put(InternalRegion region, DiskEntry entry, ValueWrapper value, boolean async)
      throws RegionClearedException {
    DiskRegion dr = region.getDiskRegion();
    DiskId id = entry.getDiskId();
    long start = async ? getStats().startFlush() : getStats().startWrite();
    if (!async) {
      dr.getStats().startWrite();
    }
    try {
      if (!async) {
        acquireReadLock(dr);
      }
      try {
        if (dr.isRegionClosed()) {
          region.getCancelCriterion().checkCancelInProgress(null);
          throw new RegionDestroyedException(
              "The DiskRegion has been closed or destroyed",
              dr.getName());
        }

        // Asif TODO: Should the htree reference in
        // DiskRegion/DiskRegion be made
        // volatile.Will theacquireReadLock ensure variable update?
        boolean doingCreate = dr.isBackup() && id.getKeyId() == INVALID_ID;
        // the call to newOplogEntryId moved down into Oplog.basicCreate
        boolean goahead = true;
        if (dr.didClearCountChange()) {
          // mbid: if the reference has changed (by a clear)
          // after a put has been made in the region
          // then we need to confirm if this key still exists in the region
          // before writing to disk
          goahead = region.basicGetEntry(entry.getKey()) == entry;
        }
        if (goahead) {
          // in overflow only mode, no need to write the key and the
          // extra data, hence if it is overflow only mode then use
          // modify and not create
          OplogSet oplogSet = getOplogSet(dr);
          if (doingCreate) {
            oplogSet.create(region, entry, value, async);
          } else {
            oplogSet.modify(region, entry, value, async);
          }
        } else {
          throw new RegionClearedException(
              String.format(
                  "Clear operation aborting the ongoing Entry %s operation for Entry with DiskId, %s",
                  ((doingCreate) ? "creation" : "modification"), id));
        }
      } finally {
        if (!async) {
          releaseReadLock(dr);
        }
      }
    } finally {
      if (async) {
        getStats().endFlush(start);
      } else {
        dr.getStats().endWrite(start, getStats().endWrite(start));
        dr.getStats().incWrittenBytes(id.getValueLength());
      }
    }
  }

  public void putVersionTagOnly(InternalRegion region, VersionTag tag, boolean async) {
    DiskRegion dr = region.getDiskRegion();
    // this method will only be called by backup oplog
    assert dr.isBackup();

    if (!async) {
      acquireReadLock(dr);
    }
    try {
      if (dr.isRegionClosed()) {
        region.getCancelCriterion().checkCancelInProgress(null);
        throw new RegionDestroyedException(
            "The DiskRegion has been closed or destroyed",
            dr.getName());
      }

      if (dr.getRegionVersionVector().contains(tag.getMemberID(), tag.getRegionVersion())) {
        // No need to write the conflicting tag to disk if the disk RVV already
        // contains this tag.
        return;
      }

      PersistentOplogSet oplogSet = getPersistentOplogSet(dr);

      oplogSet.getChild().saveConflictVersionTag(region, tag, async);
    } finally {
      if (!async) {
        releaseReadLock(dr);
      }
    }
  }

  /**
   * Returns the value of the key/value pair with the given diskId. Updates all of the necessary
   * {@linkplain DiskRegionStats statistics}
   *
   */
  Object get(DiskRegion dr, DiskId id) {
    acquireReadLock(dr);
    try {
      int count = 0;
      RuntimeException ex = null;
      while (count < 3) {
        // retry at most 3 times
        BytesAndBits bb = null;
        try {
          if (dr.isRegionClosed()) {
            throw new RegionDestroyedException(
