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

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Collections;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.persistence.DiskStoreID;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * RegionVersionVector tracks the highest region-level version number of operations applied to a
 * region for each member that has the region.
 * <p>
 *
 */
public abstract class RegionVersionVector<T extends VersionSource<?>>
    implements DataSerializableFixedID, MembershipListener {

  private static final Logger logger = LogService.getLogger();

  // TODO:LOG:CONVERT: REMOVE THIS
  public static final boolean DEBUG =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "VersionVector.VERBOSE");



  //////////////////// The following statics exist for unit testing. ////////////////////////////

  /** maximum ms wait time while waiting for dominance to be achieved */
  public static final long MAX_DOMINANCE_WAIT_TIME =
      Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "max-dominance-wait-time", 5000);

  /** maximum ms pause time while waiting for dominance to be achieved */
  public static final long DOMINANCE_PAUSE_TIME =
      Math.min(Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "dominance-pause-time", 300),
          MAX_DOMINANCE_WAIT_TIME);

  private static final int INITIAL_CAPACITY = 2;
  private static final int CONCURRENCY_LEVEL = 2;
  private static final float LOAD_FACTOR = 0.75f;
  ////////////////////////////////////////////////////////////////////////////////////////////////



  /** map of member to version h older. This is the actual version "vector" */
  private final ConcurrentHashMap<T, RegionVersionHolder<T>> memberToVersion;

  /** current version in the local region for generating next version */
  private final AtomicLong localVersion = new AtomicLong(0);
  /**
   * The list of exceptions for the local member. The version held in this RegionVersionHolder may
   * not be accurate, but the exception list is. We can have exceptions for our own id if we recover
   * from disk or GII from a peer that has exceptions from us.
   *
   * The version held in this object can lag behind the localVersion atomic long, because that long
   * is incremented without obtaining a lock. Operations that use the localException list are
   * responsible for updating the version of the local exceptions under lock.
   */
  private RegionVersionHolder<T> localExceptions;

  /** highest reaped tombstone region-version for this member */
  private final AtomicLong localGCVersion = new AtomicLong(0);

  /** the member that this version vector applies to */
  private T myId;



  /**
   * a flag stating whether this vector contains only the version information for a single member.
   * This is used when a member crashed to transmit only the version information for that member.
   */
  private boolean singleMember;

  /** a flag to prevent accidental serialization of a live member */
  private transient boolean isLiveVector;

  private transient LocalRegion region;

  private final ConcurrentHashMap<T, Long> memberToGCVersion;

  /** map of canonical IDs for this RVV that are not in the memberToVersion map */
  @SuppressWarnings("unchecked")
  private transient Map<T, T> canonicalIds = Collections.EMPTY_MAP;

  private final Object canonicalIdLock = new Object();

  /** is recording disabled? */
  private transient boolean recordingDisabled;

  /** is this a vector in a client cache? */
  private transient boolean clientVector;

  /**
   * this read/write lock is used to stop generation of new versions by the vector while a
   * region-level operation is underway. The locking scheme assumes that only one region-level RVV
   * operation is allowed at a time on a region across the distributed system. If that changes then
   * the locking scheme here may need additional work.
   */
  private final transient ReentrantReadWriteLock versionLock = new ReentrantReadWriteLock();
  private transient volatile boolean locked; // this is only modified by the version locking thread
  private transient volatile boolean doUnlock; // this is only modified by the version locking
                                               // thread
  private transient InternalDistributedMember lockOwner; // guarded by lockWaitSync

  private final transient Object clearLockSync = new Object(); // sync for coordinating thread
                                                               // startup and lockOwner setting

  /**
   * constructor used to create a cloned vector
   */
  protected RegionVersionVector(T ownerId, ConcurrentHashMap<T, RegionVersionHolder<T>> vector,
      long version, ConcurrentHashMap<T, Long> gcVersions, long gcVersion, boolean singleMember,
      RegionVersionHolder<T> localExceptions) {
    myId = ownerId;
    memberToVersion = vector;
    memberToGCVersion = gcVersions;
    localGCVersion.set(gcVersion);
    localVersion.set(version);
    this.singleMember = singleMember;
    this.localExceptions = localExceptions;
  }

  /**
   * deserialize a cloned vector
   */
  public RegionVersionVector() {
    memberToVersion = new ConcurrentHashMap<>(INITIAL_CAPACITY,
        LOAD_FACTOR, CONCURRENCY_LEVEL);
    memberToGCVersion =
        new ConcurrentHashMap<>(INITIAL_CAPACITY, LOAD_FACTOR, CONCURRENCY_LEVEL);
  }

  /**
   * create a live version vector for a region
   */
  public RegionVersionVector(T ownerId) {
    this(ownerId, null);
  }

  /**
   * create a live version vector for a region
   */
  public RegionVersionVector(T ownerId, LocalRegion owner) {
    this(ownerId, owner, 0);
  }

  @VisibleForTesting
  RegionVersionVector(T ownerId, LocalRegion owner, long version) {
    myId = ownerId;
    isLiveVector = true;
    region = owner;
    localExceptions = new RegionVersionHolder<>(0);
    localExceptions.id = myId;
    memberToVersion =
        new ConcurrentHashMap<>(INITIAL_CAPACITY, LOAD_FACTOR, CONCURRENCY_LEVEL);
    memberToGCVersion =
        new ConcurrentHashMap<>(INITIAL_CAPACITY, LOAD_FACTOR, CONCURRENCY_LEVEL);
    localVersion.set(version);
  }

  /**
   * Retrieve a vector that can be sent to another member. This clones all of the version
   * information to protect against concurrent modification during serialization
   */
  public RegionVersionVector<T> getCloneForTransmission() {
    Map<T, RegionVersionHolder<T>> liveHolders;
    liveHolders = new HashMap<>(memberToVersion);
    ConcurrentHashMap<T, RegionVersionHolder<T>> clonedHolders =
        new ConcurrentHashMap<>(liveHolders.size(), LOAD_FACTOR,
            CONCURRENCY_LEVEL);
    for (Map.Entry<T, RegionVersionHolder<T>> entry : liveHolders.entrySet()) {
      clonedHolders.put(entry.getKey(), entry.getValue().clone());
    }
    ConcurrentHashMap<T, Long> gcVersions = new ConcurrentHashMap<>(
        memberToGCVersion.size(), LOAD_FACTOR, CONCURRENCY_LEVEL);
    gcVersions.putAll(memberToGCVersion);
    RegionVersionHolder<T> clonedLocalHolder;
    clonedLocalHolder = localExceptions.clone();
    // Make sure the holder that we send to the peer does
    // have an accurate RegionVersionHolder for our local version
    return createCopy(myId, clonedHolders, localVersion.get(), gcVersions,
        localGCVersion.get(), false, clonedLocalHolder);
  }

  protected abstract RegionVersionVector<T> createCopy(T ownerId,
      ConcurrentHashMap<T, RegionVersionHolder<T>> vector, long version,
      ConcurrentHashMap<T, Long> gcVersions, long gcVersion, boolean singleMember,
      RegionVersionHolder<T> clonedLocalHolder);


  /**
   * Retrieve a vector that can be sent to another member. This clones only the version information
   * for the given ID.
   * <p>
   * The clone returned by this method does not have distributed garbage-collection information.
   */
  public RegionVersionVector<T> getCloneForTransmission(T mbr) {
    Map<T, RegionVersionHolder<T>> liveHolders;
    liveHolders = new HashMap<>(memberToVersion);
    RegionVersionHolder<T> holder = liveHolders.get(mbr);
    if (holder == null) {
      if (mbr.isDiskStoreId() && mbr.equals(myId)) {
        // For region recovered from disk, we may have local exceptions needs to be
        // brought back during region synchronization
        holder = localExceptions.clone();
        holder.setVersion(localVersion.get());
      } else {
        holder = new RegionVersionHolder<>(-1);
      }
    } else {
      holder = holder.clone();
    }
    return createCopy(myId,
        new ConcurrentHashMap<>(Collections.singletonMap(mbr, holder)), 0,
        new ConcurrentHashMap<>(INITIAL_CAPACITY, LOAD_FACTOR, CONCURRENCY_LEVEL), 0, true,
        new RegionVersionHolder<>(-1));
  }


  /**
   * Retrieve a collection of tombstone GC region-versions
   */
  public Map<T, Long> getTombstoneGCVector() {
    Map<T, Long> result;
    synchronized (memberToGCVersion) {
      result = new HashMap<>(memberToGCVersion);
    }
    if (localGCVersion.get() != 0) {
      result.put(myId, localGCVersion.get());
    }
    return result;
  }


  /** returns true if all of the GC versions in the given map have already been processed here */
  public boolean containsTombstoneGCVersions(Map<T, Long> regionGCVersions) {
    Long myVersion = regionGCVersions.get(myId);
    if (myVersion != null) {
      if (localGCVersion.get() < myVersion) {
        return false;
      }
    }
    synchronized (memberToGCVersion) {
      for (Map.Entry<T, Long> entry : regionGCVersions.entrySet()) {
        Long version = memberToGCVersion.get(entry.getKey());
        if (version == null || version < entry.getValue()) {
          return false;
        }
      }
    }
    return true;
  }

  /**
   * locks against new version generation and returns the current region version number
   *
   */
  public long lockForClear(String regionPath, DistributionManager dm,
      InternalDistributedMember locker) {
    lockVersionGeneration(regionPath, dm, locker);
    return localVersion.get();
  }

  /** unlocks version generation for clear() operations */
  public void unlockForClear(InternalDistributedMember locker) {
    synchronized (clearLockSync) {
      if (logger.isDebugEnabled()) {
        logger.debug("Unlocking for clear, from member {} RVV {}", locker,
            System.identityHashCode(this));
      }
      if (lockOwner != null && !locker.equals(lockOwner)) {
        if (logger.isDebugEnabled()) {
          logger.debug("current clear lock owner was {} not unlocking", lockOwner);
        }
        // this method is invoked by memberDeparted events and may not be for the current lock owner
        return;
      }
      unlockVersionGeneration(locker);
    }
  }

  /**
   * This schedules a thread that owns the version-generation write-lock for this vector. The method
   * unlockVersionGeneration notifies the thread to release the lock and terminate its run.
   *
   * @param dm the distribution manager - used to obtain an executor to hold the thread
   * @param locker the member requesting the lock (currently not used)
   */
  private void lockVersionGeneration(final String regionPath, final DistributionManager dm,
      final InternalDistributedMember locker) {
    final CountDownLatch acquiredLock = new CountDownLatch(1);
    if (logger.isDebugEnabled()) {
      logger.debug("Locking version generation for {} region {} RVV {}", locker, regionPath,
          System.identityHashCode(this));
    }
    // this could block for a while if a limit has been set on the waiting-thread-pool
    dm.getExecutors().getWaitingThreadPool().execute(new Runnable() {
      @Override
      @edu.umd.cs.findbugs.annotations.SuppressWarnings(
          value = {"UL_UNRELEASED_LOCK", "IMSE_DONT_CATCH_IMSE"})
      public void run() {
        boolean haveLock = false;
        synchronized (clearLockSync) {
          try {
            // TODO Following code does not seem necessary as dlock has been taken
            // so no two threads will try to enter in this code section.
            while (locked && dm.isCurrentMember(locker)) {
              try {
                clearLockSync.wait();
              } catch (InterruptedException e) {
                // okay to ignore - release the lock and exit
              }
            }
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "Waiting thread is now locking version generation for {} region {} RVV {}",
                  locker, regionPath, System.identityHashCode(this));
            }
            try {
              versionLock.writeLock().lock();
              lockOwner = locker;
              doUnlock = false;
              locked = true;
              haveLock = true;
              acquiredLock.countDown();
            } catch (IllegalMonitorStateException e) {
              // dlock on the clear() operation should prevent this from happening
              logger.fatal(
                  "Request from {} to block operations found that operations are already blocked by member {}.",
                  new Object[] {locker, lockOwner});
              return;
            }

            while (!doUnlock && dm.isCurrentMember(locker)) {
              try {
                clearLockSync.wait(250);
              } catch (InterruptedException e) {
                // okay to ignore - release the lock and exit
              }
            }
          } finally {
            if (haveLock) {
              locked = false; // this must be clear when the writeLock is released
                              // so we don't get warnings about it still being locked
              versionLock.writeLock().unlock();
              doUnlock = false;
              clearLockSync.notifyAll();
              // leave lockOwner set so we can see who the last lock request came from
            }
            acquiredLock.countDown();
          }
        }
      }
    });
    boolean interrupted = false;
    while (dm.isCurrentMember(locker)) {
      try {
        if (acquiredLock.await(250, TimeUnit.MILLISECONDS)) {
          break;
        }
      } catch (InterruptedException e) {
        interrupted = true;
      }
    }
    if (interrupted) {
      Thread.currentThread().interrupt();
    }
    if (logger.isDebugEnabled()) {
      logger.debug("Done locking");
    }

  }

  private void unlockVersionGeneration(final InternalDistributedMember locker) {
    synchronized (clearLockSync) {
      doUnlock = true;
      clearLockSync.notifyAll();
    }
  }

  /**
   * return the next local version number
   */
  public long getNextVersion() {
    return getNextVersion(true);
  }

  /**
   * return the next local version number for a clear() operation, bypassing lock checks
   */
  public long getNextVersionWhileLocked() {
    return getNextVersion(false);
  }

  /**
   * return the next local version number
   */
  private long getNextVersion(boolean checkLocked) {
    if (checkLocked && locked) {
      // this should never be the case. If version generation is locked and we get here
      // then the path to this point is not protected by getting the version generation
      // lock from the RVV but it should be
      if (logger.isDebugEnabled()) {
        logger.debug("Generating a version tag when version generation is locked by {}",
            lockOwner);
      }
    }
    long new_version = localVersion.incrementAndGet();
    // since there could be special exception, we have to use recordVersion()
    recordVersion(getOwnerId(), new_version);
    return new_version;
  }

  /** obtain a lock to prevent concurrent clear() from happening */
  public void lockForCacheModification(LocalRegion owner) {
    if (owner.getServerProxy() == null) {
      versionLock.readLock().lock();
    }
  }

  /** release the lock preventing concurrent clear() from happening */
  public void releaseCacheModificationLock(LocalRegion owner) {
    if (owner.getServerProxy() == null) {
      versionLock.readLock().unlock();
    }
  }

  /** obtain a lock to prevent concurrent clear() from happening */
  public void lockForCacheModification() {
    versionLock.readLock().lock();
  }

  /** release the lock preventing concurrent clear() from happening */
  public void releaseCacheModificationLock() {
    versionLock.readLock().unlock();
  }

  private void syncLocalVersion() {
    long v = localVersion.get();
    synchronized (localExceptions) {
      if (v != localExceptions.version) {
        if (logger.isDebugEnabled()) {
          logger.debug("Adjust localExceptions.version {} to equal localVersion {}",
              localExceptions.version, localVersion.get());
        }
        localExceptions.version = v;
      }
    }
  }

  /**
   * return the current version for this member
   */
  public long getCurrentVersion() {
    synchronized (localExceptions) {
      syncLocalVersion();
      return localExceptions.getVersion();
    }
  }

  /**
   * return the current version for this member
   */
  public RegionVersionHolder<T> getLocalExceptions() {
    return localExceptions;
  }

  /**
   * return version holder for this member
   */
  public RegionVersionHolder<T> getHolderForMember(T id) {
    if (id.equals(myId)) {
      return localExceptions;
    } else {
      return memberToVersion.get(id);
    }
  }

  /**
   * returns the ID of the member that owns this version vector
   */
  public T getOwnerId() {
    return myId;
  }


  /**
   * turns off recording of versions for this vector. This can be used when recording of versions is
   * not necessary, as in an empty region
   */
  public void turnOffRecordingForEmptyRegion() {
    recordingDisabled = true;
  }

  /**
   * client version vectors only record GC numbers and don't keep exceptions, etc, because there
   * could be MANY of them
   */
  public void setIsClientVector() {
    clientVector = true;
  }

  /**
   * record all of the version information from an initial image provider
   */
  public void recordVersions(RegionVersionVector<T> otherVector) {
    synchronized (memberToVersion) {
      for (Map.Entry<T, RegionVersionHolder<T>> entry : otherVector.getMemberToVersion()
          .entrySet()) {
        T mbr = entry.getKey();
        RegionVersionHolder<T> otherHolder = entry.getValue();

        initializeVersionHolder(mbr, otherHolder);
      }
      // Get the set of local exceptions from the other vector
      // before directly accessing localExceptions, should sync its this.version with localVersion
      otherVector.syncLocalVersion();
      initializeVersionHolder(otherVector.getOwnerId(), otherVector.localExceptions);

      if (otherVector.getCurrentVersion() > 0
          && !memberToVersion.containsKey(otherVector.getOwnerId())) {
        recordVersion(otherVector.getOwnerId(), otherVector.getCurrentVersion());
      }

      // check if I have updates from members that the otherVector does not have
      // If yes, these are unfinished ops and should be cleaned
      for (T mbr : memberToVersion.keySet()) {
        if (!otherVector.memberToVersion.containsKey(mbr)
            && !mbr.equals(otherVector.getOwnerId())) {
          RegionVersionHolder holder = memberToVersion.get(mbr);
          initializeVersionHolder(mbr, new RegionVersionHolder(0));
        }
      }
      if (!otherVector.memberToVersion.containsKey(myId)
          && !myId.equals(otherVector.getOwnerId())) {
        initializeVersionHolder(myId, new RegionVersionHolder(0));
      }

      synchronized (memberToGCVersion) {
        for (Map.Entry<T, Long> entry : otherVector.getMemberToGCVersion().entrySet()) {
          T member = entry.getKey();
          Long value = entry.getValue();
          if (member.equals(myId)) {
            // If this entry is for our id, update our local GC version
            long currentValue;
            while ((currentValue = localGCVersion.get()) < value) {
              localGCVersion.compareAndSet(currentValue, value);
            }
          } else {
            // Update the memberToGCVersionMap.
            Long myVersion = memberToGCVersion.get(entry.getKey());
            if (myVersion == null || myVersion < entry.getValue()) {
              memberToGCVersion.put(entry.getKey(), entry.getValue());
            }
          }
        }
      }
    }
  }

  public void initializeVersionHolder(T mbr, RegionVersionHolder<T> otherHolder) {
    RegionVersionHolder<T> h = memberToVersion.get(mbr);
    if (h == null) {
      if (!mbr.equals(myId)) {
        h = otherHolder.clone();
        h.makeReadyForRecording();
        if (h.id == null) {
          h.id = mbr;
        }
        memberToVersion.put(mbr, h);
      } else {
        RegionVersionHolder<T> vh = otherHolder;
        long version = vh.version;
        updateLocalVersion(version);
        localExceptions.initializeFrom(vh);
      }
    } else {
      // holders must be modified under synchronization
      h.initializeFrom(otherHolder);
    }
  }

  void updateLocalVersion(long newVersion) {
    boolean needToTrySetAgain;
    do {
      needToTrySetAgain = false;
      long currentVersion = localVersion.get();
      if (currentVersion < newVersion) {
        needToTrySetAgain = !compareAndSetVersion(currentVersion, newVersion);
      }
    } while (needToTrySetAgain);
  }

  boolean compareAndSetVersion(long currentVersion, long newVersion) {
    return localVersion.compareAndSet(currentVersion, newVersion);
  }

  /**
   * Records a received region-version. These are transmitted in VersionTags in messages between
   * peers and from servers to clients.
   *
   * @param tag the version information
   */
  public void recordVersion(T mbr, VersionTag<T> tag) {
    tag.setRecorded();
    assert tag.isRecorded();
    T member = tag.getMemberID();
    if (member == null) {
      member = mbr;
    }

    if (myId.equals(member)) {
      // We can be asked to record a version for the local member if a persistent
      // member is restarted and an event is replayed after the persistent member
      // recovers. So we can only assert that the local member has already seen
      // the replayed event.
      synchronized (localExceptions) {
        if (localVersion.get() < tag.getRegionVersion() && region != null
            && region.isInitialized() && region.getDataPolicy().withPersistence()) {
          Assert.fail(
              "recordVersion invoked for a local version tag that is higher than our local version. rvv="
                  + this + ", tag=" + tag + " " + region.getName());
        }
      }
    }

    recordVersion(member, tag.getRegionVersion());
  }

  /**
   * Records a received region-version. These are transmitted in VersionTags in messages between
   * peers and from servers to clients. In general you should use recordVersion(mbr, versionTag) so
   * that the tag is marked as having been recorded. This will keep
   * DistributedCacheOperation.basicProcess() from trying to record it again.
   *
   * This method is also called for versions which have been recovered from disk.
   *
   * @param member the peer that performed the operation
   * @param version the version of the peers region that reflects the operation
   */
  public void recordVersion(T member, long version) {
    T mbr = member;

    if (recordingDisabled || clientVector) {
      return;
    }

    RegionVersionHolder<T> holder;

    if (mbr.equals(myId)) {
      // If we are recording a version for the local member,
      // use the local exception list.
      holder = localExceptions;

      synchronized (holder) {
        // Advance the version held in the local
        // exception list to match the atomic long
        // we using for the local version.
        holder.version = localVersion.get();
      }
      updateLocalVersion(version);
    } else {
      // Find the version holder object
      holder = memberToVersion.get(mbr);
      if (holder == null) {
        synchronized (memberToVersion) {
          // Look for the holder under lock
          holder = memberToVersion.get(mbr);
          if (holder == null) {
            mbr = getCanonicalId(mbr);
            holder = new RegionVersionHolder<>(mbr);
            memberToVersion.put(holder.id, holder);
          }
        }
      }
    }

    // Update the version holder
    if (logger.isTraceEnabled(LogMarker.RVV_VERBOSE)) {
      logger.trace(LogMarker.RVV_VERBOSE, "Recording rv{} for {}", version, mbr);
    }
    holder.recordVersion(version);
  }



  /**
   * Records a version holder that we have recovered from disk. This version holder replaces the
   * current version holder if it dominates the version holder we already have. This method will
   * called once for each oplog we recover.
   *
   */
  @edu.umd.cs.findbugs.annotations.SuppressWarnings(
      value = "ML_SYNC_ON_FIELD_TO_GUARD_CHANGING_THAT_FIELD",
      justification = "sync on localExceptions guards concurrent modification but this is a replacement")
  public void initRecoveredVersion(T member, RegionVersionHolder<T> v, boolean latestOplog) {
    RegionVersionHolder<T> recovered = v.clone();

    if (member == null || member.equals(myId)) {
      // if this is the version for the local member, update our local info

      // update the local exceptions
      synchronized (localExceptions) {
        // Fix for 45622 - We only take the version holder from the latest
        // oplog. There may be more than one RVV in the latest oplog, in which
        // case we want to end up with the last RVV from the latest oplog
        if (latestOplog || localVersion.get() == 0) {
          localExceptions = recovered;
          if (logger.isTraceEnabled(LogMarker.RVV_VERBOSE)) {
            logger.trace(LogMarker.RVV_VERBOSE, "initRecoveredVersion setting local version to {}",
                recovered.version);
          }
          localVersion.set(recovered.version);
        }
      }
    } else {
      // If this is not a local member, update the member to version map
      Long gcVersion = memberToGCVersion.get(member);

      synchronized (memberToVersion) {
        RegionVersionHolder<T> oldVersion = memberToVersion.get(member);

        // Fix for 45622 - We only take the version holder from the latest
        // oplog. There may be more than one RVV in the latest oplog, in which
        // case we want to end up with the last RVV from the latest oplog
        if (latestOplog || oldVersion == null || oldVersion.version == 0) {
          if (gcVersion != null) {
            recovered.removeExceptionsOlderThan(gcVersion);
          }
          memberToVersion.put(member, recovered);
        }
      }
    }
  }

  /**
   * get the last recorded region version number for the given member
   */
  public long getVersionForMember(T mbr) {
    RegionVersionHolder<T> holder = memberToVersion.get(mbr);
    if (holder == null) {
      if (mbr.equals(myId)) {
        return getCurrentVersion();
      } else {
        return 0;
      }
    } else {
      return holder.getVersion();
    }
  }

  /**
   * Returns a list of the members that have been marked as having left the system.
   */
  public Set<T> getDepartedMembersSet() {
    synchronized (memberToVersion) {
      Set<T> result = new HashSet<>();
      for (RegionVersionHolder<T> h : memberToVersion.values()) {
        if (h.isDepartedMember) {
          result.add(h.id);
        }
      }
      return result;
    }
  }


  /**
   * Test to see if this vector has seen the given version.
   *
   * @return true if this vector has seen the given version
   */
  public boolean contains(T id, long version) {
    RegionVersionHolder<T> holder = memberToVersion.get(id);
    // For region synchronization.
    if (isForSynchronization()) {
      if (holder == null) {
        // we only care about missing changes from a particular member, and this
        // vector is known to contain that member's version holder
        return true;
      }
      if (id.equals(myId)) {
        if (!myId.isDiskStoreId()) {
          // a sync vector only has one holder if not recovered from persistence,
          // no valid version for the vector's owner
          return true;
        }
      }
      return holder.contains(version);
    }

    // Regular GII
    if (id.equals(myId)) {
      if (getCurrentVersion() < version) {
        return false;
      } else {
        return !localExceptions.hasExceptionFor(version);
      }
    }
    if (holder == null) {
      return false;
    } else {
      return holder.contains(version);
    }
  }

  /**
   * Removes departed members not in the given collection of IDs from the version vector
   *
   * @param idsToKeep collection of the kind of IDs appropriate for this vector
   */
  public void removeOldMembers(Set<VersionSource<T>> idsToKeep) {
    synchronized (memberToVersion) {
      for (Iterator<Map.Entry<T, RegionVersionHolder<T>>> it =
          memberToVersion.entrySet().iterator(); it.hasNext();) {
        Map.Entry<T, RegionVersionHolder<T>> entry = it.next();
        if (entry.getValue().isDepartedMember) {
          if (!idsToKeep.contains(entry.getKey())) {
            it.remove();
            memberToGCVersion.remove(entry.getKey());
            synchronized (canonicalIdLock) {
              canonicalIds.remove(entry.getKey());
            }
          }
        }
      }
    }
  }



  /**
   * Test hook - does this vector hold an entry for the given ID?
   */
  public boolean containsMember(T id) {
    if (memberToVersion.containsKey(id)) {
      return true;
    }
    if (memberToGCVersion.containsKey(id)) {
      return true;
    }
    return canonicalIds.containsKey(id);
  }

  /**
   * This marks the given entry as departed, making it eligible to be removed during an operation
   * like DistributedRegion.synchronizeWith()
   *
   */
  protected void markDepartedMember(T id) {
    synchronized (memberToVersion) {
      RegionVersionHolder<T> holder = memberToVersion.get(id);
      if (holder != null) {
        holder.isDepartedMember = true;
      }
    }
  }

  /**
   * check to see if tombstone removal in this RVV indicates that tombstones have been removed from
   * its Region that have not been removed from the argument's Region. If this is the case, then a
   * delta GII may leave entries in the other RVV's Region that should be deleted.
   *
   * @return true if there have been tombstone removals in this vector's Region that were not done
   *         in the argument's region
   */
