/*
 * 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.
 */
/*
 * ConcurrentHashMap implementation adapted from JSR 166 backport
 * (http://backport-jsr166.sourceforge.net) JDK5 version release 3.1 with modifications to use
 * generics where appropriate: backport-util-concurrent-Java60-3.1-src.tar.gz
 *
 * Primary change is to allow HashEntry be an interface so that custom HashEntry implementations can
 * be plugged in. These HashEntry objects are now assumed to be immutable in the sense that they
 * cannot and should not be cloned in a rehash, and the rehash mechanism has been recoded using
 * locking for that. For Geode, this is now used to plugin the RegionEntry implementation directly
 * as a HashEntry instead of having it as a value and then HashEntry as a separate object having
 * references to key/value which reduces the entry overhead substantially. Other change is to add a
 * "create" method that creates a new object using the {@link MapCallback} interface only if
 * required unlike "putIfAbsent" that requires a pre-built object that may ultimately be thrown
 * away. Also added a "removeConditionally" method that allows for evaluation of an arbitrary
 * condition before removal from the map (unlike the normal "remove" that can only check for
 * equality with a provided object). In addition, the segments are now locked using read-write
 * locks. File has been reformatted to conform to GemStone conventions. GemStone additions have been
 * marked with "GemStone addition". GemStone changes have been marked with "GemStone change(s)".
 *
 * Original license follows below.
 */
/*
 * Written by Doug Lea with assistance from members of JCP JSR-166 Expert Group and released to the
 * public domain, as explained at http://creativecommons.org/licenses/publicdomain
 */
package org.apache.geode.internal.util.concurrent;

import java.io.IOException;
import java.io.Serializable;
import java.util.AbstractCollection;
import java.util.AbstractMap;
import java.util.AbstractSet;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Enumeration;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.geode.CancelException;
import org.apache.geode.internal.cache.RegionEntry;
import org.apache.geode.internal.cache.entries.OffHeapRegionEntry;
import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
import org.apache.geode.internal.offheap.OffHeapClearRequired;
import org.apache.geode.internal.size.SingleObjectSizer;
import org.apache.geode.internal.util.ArrayUtils;
import org.apache.geode.logging.internal.executors.LoggingThread;

/**
 * A hash table supporting full concurrency of retrievals and adjustable expected concurrency for
 * updates. This class obeys the same functional specification as {@link java.util.Hashtable}, and
 * includes versions of methods corresponding to each method of <tt>Hashtable</tt>. However, even
 * though all operations are thread-safe, retrieval operations do <em>not</em> entail locking, and
 * there is <em>not</em> any support for locking the entire table in a way that prevents all access.
 * This class is fully interoperable with <tt>Hashtable</tt> in programs that rely on its thread
 * safety but not on its synchronization details.
 *
 * <p>
 * Retrieval operations (including <tt>get</tt>) generally do not block, so may overlap with update
 * operations (including <tt>put</tt> and <tt>remove</tt>). Retrievals reflect the results of the
 * most recently <em>completed</em> update operations holding upon their onset. For aggregate
 * operations such as <tt>putAll</tt> and <tt>clear</tt>, concurrent retrievals may reflect
 * insertion or removal of only some entries. Similarly, Iterators and Enumerations return elements
 * reflecting the state of the hash table at some point at or since the creation of the
 * iterator/enumeration. They do <em>not</em> throw
 * {@link java.util.ConcurrentModificationException}. However, iterators are designed to be used by
 * only one thread at a time.
 *
 * <p>
 * The allowed concurrency among update operations is guided by the optional
 * <tt>concurrencyLevel</tt> constructor argument (default <tt>16</tt>), which is used as a hint for
 * internal sizing. The table is internally partitioned to try to permit the indicated number of
 * concurrent updates without contention. Because placement in hash tables is essentially random,
 * the actual concurrency will vary. Ideally, you should choose a value to accommodate as many
 * threads as will ever concurrently modify the table. Using a significantly higher value than you
 * need can waste space and time, and a significantly lower value can lead to thread contention. But
 * overestimates and underestimates within an order of magnitude do not usually have much noticeable
 * impact. A value of one is appropriate when it is known that only one thread will modify and all
 * others will only read. Also, resizing this or any other kind of hash table is a relatively slow
 * operation, so, when possible, it is a good idea to provide estimates of expected table sizes in
 * constructors.
 *
 * <p>
 * This class and its views and iterators implement all of the <em>optional</em> methods of the
 * {@link Map} and {@link Iterator} interfaces.
 *
 * <p>
 * Like {@link java.util.Hashtable} but unlike {@link java.util.HashMap}, this class does
 * <em>not</em> allow <tt>null</tt> to be used as a key or value.
 *
 * <p>
 * This class is a member of the <a href="{@docRoot} /../technotes/guides/collections/index.html">
 * Java Collections Framework</a>.
 *
 * <p>
 * Modifications made to the Java {@link java.util.Hashtable} implementation
 * are indicated using comments to mark the beginning and end of Geode changes.
 *
 * @since Java 1.5
 * @author Doug Lea
 * @param <K> the type of keys maintained by this map
 * @param <V> the type of mapped values
 */
public class CustomEntryConcurrentHashMap<K, V> extends AbstractMap<K, V>
    implements ConcurrentMapWithReusableEntries<K, V>, Serializable {

  private static final long serialVersionUID = -7056732555635108300L;

  /*
   * The basic strategy is to subdivide the table among Segments, each of which itself is a
   * concurrently readable hash table.
   */

  /* ---------------- Constants -------------- */

  /**
   * The default initial capacity for this table, used when not otherwise specified in a
   * constructor.
   */
  public static final int DEFAULT_INITIAL_CAPACITY = 16;

  /**
   * The default load factor for this table, used when not otherwise specified in a constructor.
   */
  public static final float DEFAULT_LOAD_FACTOR = 0.75f;

  /**
   * The default concurrency level for this table, used when not otherwise specified in a
   * constructor.
   */
  public static final int DEFAULT_CONCURRENCY_LEVEL = 16;

  /**
   * The maximum capacity, used if a higher value is implicitly specified by either of the
   * constructors with arguments. MUST be a power of two <= 1<<30 to ensure that entries are
   * indexable using ints.
   */
  static final int MAXIMUM_CAPACITY = 1 << 30;

  /**
   * The maximum number of segments to allow; used to bound constructor arguments.
   */
  static final int MAX_SEGMENTS = 1 << 16; // slightly conservative

  /**
   * Number of unsynchronized retries in size and containsValue methods before resorting to locking.
   * This is used to avoid unbounded retries if tables undergo continuous modification which would
   * make it impossible to obtain an accurate result.
   */
  static final int RETRIES_BEFORE_LOCK = 2;

  // Geode addition
  /**
   * Token object to indicate that {@link #remove(Object)} does not need to compare against provided
   * value before removing from segment.
   */
  private static final Object NO_OBJECT_TOKEN = new Object();

  // End Geode addition

  /* ---------------- Fields -------------- */

  /**
   * Mask value for indexing into segments. The upper bits of a key's hash code are used to choose
   * the segment.
   */
  final int segmentMask;

  /**
   * Shift value for indexing within segments.
   */
  final int segmentShift;

  /**
   * The segments, each of which is a specialized hash table
   */
  final Segment<K, V>[] segments;

  /**
   * {@link HashEntryCreator} for the map to create {@link HashEntry}s.
   */
  final HashEntryCreator<K, V> entryCreator;

  /**
   * If true then use equals for comparing key and value equality else use reference-equality like
   * an {@link IdentityHashMap}.
   */
  final boolean compareValues;

  transient Set<K> keySet;
  transient Set<Map.Entry<K, V>> entrySet;
  transient Set<Map.Entry<K, V>> reusableEntrySet; // Geode addition
  transient Collection<V> values;

  /* ---------------- Small Utilities -------------- */

  /**
   * Applies a supplemental hash function to a given hashCode, which defends against poor quality
   * hash functions. This is critical because ConcurrentHashMap uses power-of-two length hash
   * tables, that otherwise encounter collisions for hashCodes that do not differ in lower or upper
   * bits.
   */
  public static int keyHash(final Object o, final boolean compareValues) {
    int h = compareValues ? o.hashCode() : System.identityHashCode(o);
    // Spread bits to regularize both segment and index locations,
    // using variant of single-word Wang/Jenkins hash.
    h += (h << 15) ^ 0xffffcd7d;
    h ^= (h >>> 10);
    h += (h << 3);
    h ^= (h >>> 6);
    h += (h << 2) + (h << 14);
    return h ^ (h >>> 16);
  }

  /**
   * Returns the segment that should be used for key with given hash
   *
   * @param hash the hash code for the key
   * @return the segment
   */
  Segment<K, V> segmentFor(final int hash) {
    if (segmentMask == 0) {
      return segments[0];
    }
    return segments[(hash >>> segmentShift) & segmentMask];
  }

  /* ---------------- Inner Classes -------------- */

  // Geode addition
  // Geode changed HashEntry to be an interface with original HashEntry
  // as the default implementation HashEntryImpl.

  /**
   * [sumedh] Interface for ConcurrentHashMap list entry. Note that this is never exported out as a
   * user-visible Map.Entry.
   *
   * Made this public so RegionEntries can directly implement this to reduce memory overhead of
   * separate
   * {@link org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntry}
   * objects for each entry in the map.
   */
  public interface HashEntry<K, V> {

    /**
     * Get the key object for this entry.
     */
    K getKey();

    /**
     * Return true if the entry's key is equal to k. GemFire addition to deal with inline keys.
     */
    boolean isKeyEqual(Object k);

    /**
     * Get the value for this entry.
     */
    V getMapValue();

    /**
     * Set the value for this entry.
     */
    void setMapValue(V newValue);

    /**
     * Get the hash value for this entry.
     */
    int getEntryHash();

    /**
     * Get the next entry, if any, in the linear chain.
     */
    HashEntry<K, V> getNextEntry();

    /**
     * Set the next entry in the linear chain.
     */
    void setNextEntry(HashEntry<K, V> n);
  }

  /**
   * ConcurrentHashMap list entry. Note that this is never exported out as a user-visible Map.Entry.
   *
   * Because the value field is volatile, not final, it is legal wrt the Java Memory Model for an
   * unsynchronized reader to see null instead of initial value when read via a data race. Although
   * a reordering leading to this is not likely to ever actually occur, the
   * Segment.readValueUnderLock method is used as a backup in case a null (pre-initialized) value is
   * ever seen in an unsynchronized access method.
   */
  static class HashEntryImpl<K, V> implements HashEntry<K, V> {

    protected final K key;

    protected final int hash;

    protected volatile V value;

    protected HashEntry<K, V> next;

    private final HashEntry<K, V> wrappedEntry;

    HashEntryImpl(final K key, final int hash, final HashEntry<K, V> next, final V value,
        final HashEntry<K, V> wrappedEntry) {
      this.key = key;
      this.hash = hash;
      this.next = next;
      this.value = value;
      this.wrappedEntry = wrappedEntry;
    }

    /**
     * @see CustomEntryConcurrentHashMap.HashEntry#getKey()
     */
    @Override
    public K getKey() {
      return key;
    }

    /**
     * @see CustomEntryConcurrentHashMap.HashEntry#getMapValue()
     */
    @Override
    public V getMapValue() {
      return value;
    }

    /**
     * @see CustomEntryConcurrentHashMap.HashEntry#setMapValue(Object)
     */
    @Override
    public void setMapValue(V newValue) {
      value = newValue;
    }

    /**
     * @see CustomEntryConcurrentHashMap.HashEntry#getEntryHash()
     */
    @Override
    public int getEntryHash() {
      return hash;
    }

    /**
     * @see CustomEntryConcurrentHashMap.HashEntry#getNextEntry()
     */
    @Override
    public HashEntry<K, V> getNextEntry() {
      return next;
    }

    /**
     * @see CustomEntryConcurrentHashMap.HashEntry#setNextEntry
     */
    @Override
    public void setNextEntry(final HashEntry<K, V> n) {
      next = n;
    }

    @Override
    public boolean isKeyEqual(Object k) {
      return k.equals(getKey());
    }
  }

  /**
   * Interface to enable creation of new
   * {@link org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntry}
   * objects by caller. This can be used, for example, to return GemFire RegionEntries directly.
   */
  public interface HashEntryCreator<K, V> {

    /**
     * Create a new
     * {@link org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntry}
     * given the key, hash, value and next element.
     */
    HashEntry<K, V> newEntry(K key, int hash, HashEntry<K, V> next, V value);

    /**
     * Get the hashCode for given key object.
     */
    int keyHashCode(Object key, boolean compareValues);
  }

  // End Geode addition

  /**
   * Segments are specialized versions of hash tables. This subclasses from ReentrantLock
   * opportunistically, just to simplify some locking and avoid separate construction.
   */
  static class Segment<K, V> extends ReentrantReadWriteLock implements Serializable {

    /*
     * Segments maintain a table of entry lists that are ALWAYS kept in a consistent state, so can
     * be read without locking. Next fields of nodes are immutable (final). All list additions are
     * performed at the front of each bin. This makes it easy to check changes, and also fast to
     * traverse. When nodes would otherwise be changed, new nodes are created to replace them. This
     * works well for hash tables since the bin lists tend to be short. (The average length is less
     * than two for the default load factor threshold.)
     *
     * Read operations can thus proceed without locking, but rely on selected uses of volatiles to
     * ensure that completed write operations performed by other threads are noticed. For most
     * purposes, the "count" field, tracking the number of elements, serves as that volatile
     * variable ensuring visibility. This is convenient because this field needs to be read in many
     * read operations anyway:
     *
     * - All (unsynchronized) read operations must first read the "count" field, and should not look
     * at table entries if it is 0.
     *
     * - All (synchronized) write operations should write to the "count" field after structurally
     * changing any bin. The operations must not take any action that could even momentarily cause a
     * concurrent read operation to see inconsistent data. This is made easier by the nature of the
     * read operations in Map. For example, no operation can reveal that the table has grown but the
     * threshold has not yet been updated, so there are no atomicity requirements for this with
     * respect to reads.
     *
     * As a guide, all critical volatile reads and writes to the count field are marked in code
     * comments.
     */

    private static final long serialVersionUID = -6972364566212065192L;

    /**
     * The number of elements in this segment's region.
     */
    transient volatile int count;

    /**
     * Number of updates that alter the size of the table. This is used during bulk-read methods to
     * make sure they see a consistent snapshot: If modCounts change during a traversal of segments
     * computing size or checking containsValue, then we might have an inconsistent view of state so
     * (usually) must retry.
     */
    transient int modCount;

    /**
     * The table is rehashed when its size exceeds this threshold. (The value of this field is
     * always <tt>(int)(capacity *
     * loadFactor)</tt>.)
     */
    transient int threshold;

    /**
     * The per-segment table.
     */
    transient volatile HashEntry<K, V>[] table;

    /**
     * The load factor for the hash table. Even though this value is same for all segments, it is
     * replicated to avoid needing links to outer object.
     *
     * @serial
     */
    final float loadFactor;

    // Geode addition

    /**
     * {@link org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntryCreator}
     * for the map to create
     * {@link org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntry}s.
     */
    final HashEntryCreator<K, V> entryCreator;

    /**
     * Lock used when updating the
     * {@link org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntry#getNextEntry()}
     * link of an entry.
     */
    final ReentrantReadWriteLock listUpdateLock;

    // End Geode addition

    Segment(final int initialCapacity, final float lf, final HashEntryCreator<K, V> entryCreator) {
      loadFactor = lf;
      this.entryCreator = entryCreator;
      listUpdateLock = new ReentrantReadWriteLock();
      setTable(Segment.newEntryArray(initialCapacity));
    }

    @SuppressWarnings("unchecked")
    static <K, V> Segment<K, V>[] newArray(final int i) {
      return new Segment[i];
    }

    // Geode added the method below
    @SuppressWarnings("unchecked")
    static <K, V> HashEntry<K, V>[] newEntryArray(final int size) {
      return new HashEntry[size];
    }

    /**
     * Sets table to new HashEntry array. Call only while holding lock or in constructor.
     */
    void setTable(final HashEntry<K, V>[] newTable) {
      threshold = (int) (newTable.length * loadFactor);
      table = newTable;
    }

    /**
     * Returns properly casted first entry of bin for given hash.
     */
    HashEntry<K, V> getFirst(final int hash) {
      final HashEntry<K, V>[] tab = table;
      return tab[hash & (tab.length - 1)];
    }

    /**
     * Reads value field of an entry under lock. Called if value field ever appears to be null. This
     * is possible only if a compiler happens to reorder a HashEntry initialization with its table
     * assignment, which is legal under memory model but is not known to ever occur.
     */
    V readValueUnderLock(final HashEntry<K, V> e) {
      final ReentrantReadWriteLock.ReadLock readLock = super.readLock();
      readLock.lock();
      final V v = e.getMapValue();
      readLock.unlock();
      return v;
    }

    /**
     * Added for GemFire since it stores some keys inline.
     */
    protected boolean equalityKeyCompare(final Object key, final HashEntry<K, V> mapEntry) {
      return mapEntry.isKeyEqual(key);
    }

    protected boolean equalityCompare(final Object v1, final Object v2) {
      return v1.equals(v2);
    }

    protected boolean equalityCompareWithNulls(final Object key, final Object mapKey) {
      if (key != mapKey) {
        if (key != null) {
          return key.equals(mapKey);
        }
        return false;
      }
      return true;
    }

    /* Specialized implementations of map methods */

    V get(final Object key, final int hash) {
      if (count != 0) { // read-volatile
        // Geode change to acquire the read lock on list updates
        final ReentrantReadWriteLock.ReadLock listLock = listUpdateLock.readLock();
        listLock.lock();
        boolean lockAcquired = true;
        HashEntry<K, V> e = getFirst(hash);
        try {
          while (e != null) {
            if (e.getEntryHash() == hash && equalityKeyCompare(key, e)) {
              final V v = e.getMapValue();
              if (v != null) {
                return v;
              }
              listLock.unlock();
              lockAcquired = false;
              return readValueUnderLock(e); // recheck
            }
            e = e.getNextEntry();
          }
        } finally {
          if (lockAcquired) {
            listLock.unlock();
          }
        }
      }
      return null;
    }

    V getNoLock(final Object key, final int hash, final boolean lockListForRead) {
      if (count != 0) { // read-volatile
        // Geode change to acquire the read lock on list updates
        ReentrantReadWriteLock.ReadLock listLock = null;
        if (lockListForRead) {
          listLock = listUpdateLock.readLock();
          listLock.lock();
        }
        HashEntry<K, V> e = getFirst(hash);
        try {
          while (e != null) {
            if (e.getEntryHash() == hash && equalityKeyCompare(key, e)) {
              return e.getMapValue();
            }
            e = e.getNextEntry();
          }
        } finally {
          if (lockListForRead) {
            listLock.unlock();
          }
        }
      }
      return null;
    }

    boolean containsKey(final Object key, final int hash) {
      if (count != 0) { // read-volatile
        // Geode change to acquire the read lock on list updates
        final ReentrantReadWriteLock.ReadLock listLock = listUpdateLock.readLock();
        listLock.lock();
        HashEntry<K, V> e = getFirst(hash);
        try {
          while (e != null) {
            if (e.getEntryHash() == hash && equalityKeyCompare(key, e)) {
              return true;
            }
            e = e.getNextEntry();
          }
        } finally {
          listLock.unlock();
        }
      }
      return false;
    }

    boolean containsValue(final Object value) {
      if (count != 0) { // read-volatile
        // Geode change to acquire the read lock on list updates
        ReentrantReadWriteLock.ReadLock readLock = listUpdateLock.readLock();
        RETRYLOOP: for (;;) {
          readLock.lock();
          final HashEntry<K, V>[] tab = table;
          final int len = tab.length;
          for (final HashEntry<K, V> kvHashEntry : tab) {
            for (HashEntry<K, V> e = kvHashEntry; e != null; e = e.getNextEntry()) {
              V v = e.getMapValue();
              if (v == null) {
                // Geode changes BEGIN
                // go back and retry from the very start with segment read lock
                readLock.unlock();
                readLock = super.readLock();
                continue RETRYLOOP;
                /*
                 * (original code) v = readValueUnderLock(e);
                 */
                // Geode changes END
              }
              if (equalityCompare(value, v)) {
                readLock.unlock();
                return true;
              }
            }
          }
          readLock.unlock();
          return false;
        }
      }
      return false;
    }

    boolean replace(final K key, final int hash, final V oldValue, final V newValue) {
      final ReentrantReadWriteLock.WriteLock writeLock = super.writeLock();
      writeLock.lock();
      try {
        HashEntry<K, V> e = getFirst(hash);
        while (e != null && (e.getEntryHash() != hash || !equalityKeyCompare(key, e))) {
          e = e.getNextEntry();
        }

        boolean replaced = false;
        if (e != null && equalityCompare(oldValue, e.getMapValue())) {
          replaced = true;
          e.setMapValue(newValue);
        }
        return replaced;
      } finally {
        writeLock.unlock();
      }
    }

    V replace(final K key, final int hash, final V newValue) {
      final ReentrantReadWriteLock.WriteLock writeLock = super.writeLock();
      writeLock.lock();
      try {
        HashEntry<K, V> e = getFirst(hash);
        while (e != null && (e.getEntryHash() != hash || !equalityKeyCompare(key, e))) {
          e = e.getNextEntry();
        }

        V oldValue = null;
        if (e != null) {
          oldValue = e.getMapValue();
          e.setMapValue(newValue);
        }
        return oldValue;
      } finally {
        writeLock.unlock();
      }
    }

    V put(final K key, final int hash, final V value, final boolean onlyIfAbsent) {
      final ReentrantReadWriteLock.WriteLock writeLock = super.writeLock();
      writeLock.lock();
      try {
        int c = count;
        if (c++ > threshold) {
          rehash();
        }
        final HashEntry<K, V>[] tab = table;
        final int index = hash & (tab.length - 1);
        final HashEntry<K, V> first = tab[index];
        HashEntry<K, V> e = first;
        while (e != null && (e.getEntryHash() != hash || !equalityKeyCompare(key, e))) {
          e = e.getNextEntry();
        }

        final V oldValue;
        if (e != null) {
          oldValue = e.getMapValue();
          if (!onlyIfAbsent) {
            e.setMapValue(value);
          }
        } else {
          oldValue = null;
          ++modCount;
          tab[index] = entryCreator.newEntry(key, hash, first, value);
          count = c; // write-volatile
        }
        return oldValue;
      } finally {
        writeLock.unlock();
      }
    }

    // Geode additions

    <C, P> V create(final K key, final int hash, final MapCallback<K, V, C, P> valueCreator,
        final C context, final P createParams, final boolean lockForRead) {
      // TODO: This can be optimized by having a special lock implementation
      // that will allow upgrade from read to write lock atomically. This can
      // cause a deadlock if two readers try to simultaneously upgrade, so the
      // upgrade should be a tryLock that will fall back to the usual way if
      // unsuccessful. The advantage of that approach is that "equals" calls in
      // the list can be avoided completely if tryLock succeeds (i.e. presumably
      // the common case of no overlap on a segment concurrently). OTOH it will
      // not be as efficient when get succeeds without a read lock (existing
      // entry) that will not need to wait for any writers.
      if (!lockForRead) {
        final V v = getNoLock(key, hash, true);
        if (v != null) {
          return v;
        }
      } else {
        final ReentrantReadWriteLock.ReadLock readLock = super.readLock();
        readLock.lock();
        try {
          final V v = getNoLock(key, hash, false);
          if (v != null) {
            // invoke the callback before returning
            valueCreator.oldValueRead(v);
            return v;
          }
        } finally {
          readLock.unlock();
        }
      }

      final ReentrantReadWriteLock.WriteLock writeLock = super.writeLock();
      writeLock.lock();
      try {
        int c = count;
        if (c++ > threshold) {
          rehash();
        }
        final HashEntry<K, V>[] tab = table;
        final int index = hash & (tab.length - 1);
        final HashEntry<K, V> first = tab[index];
        HashEntry<K, V> e = first;
        while (e != null && (e.getEntryHash() != hash || !equalityKeyCompare(key, e))) {
          e = e.getNextEntry();
        }

        final V currentValue;
        if (e == null) {
          ++modCount;
          currentValue = valueCreator.newValue(key, context, createParams);
          tab[index] = entryCreator.newEntry(key, hash, first, currentValue);
          count = c; // write-volatile
          return currentValue;
        } else {
          currentValue = e.getMapValue();
          if (lockForRead) {
            // invoke the callback before returning an existing value
            valueCreator.oldValueRead(currentValue);
          }
          return currentValue;
        }
      } finally {
        writeLock.unlock();
      }
    }

    V get(final Object key, final int hash, final MapCallback<K, V, ?, ?> readCallback) {
      final ReentrantReadWriteLock.ReadLock readLock = super.readLock();
      readLock.lock();
      try {
        if (count != 0) { // read-volatile
          HashEntry<K, V> e = getFirst(hash);
          while (e != null) {
            if (e.getEntryHash() == hash && equalityKeyCompare(key, e)) {
              final V v = e.getMapValue();
              if (v != null) {
                if (readCallback != null) {
                  readCallback.oldValueRead(v);
                }
                return v;
              }
            }
            e = e.getNextEntry();
          }
        }
      } finally {
        readLock.unlock();
      }
      return null;
    }

    // End Geode additions

    void rehash() {
      final HashEntry<K, V>[] oldTable = table;
      final int oldCapacity = oldTable.length;
      if (oldCapacity >= MAXIMUM_CAPACITY) {
        return;
      }

      /*
       * Reclassify nodes in each list to new Map. Because we are using power-of-two expansion, the
       * elements from each bin must either stay at same index, or move with a power of two offset.
       * We eliminate unnecessary node creation by catching cases where old nodes can be reused
       * because their next fields won't change. Statistically, at the default threshold, only about
       * one-sixth of them need cloning when a table doubles. The nodes they replace will be garbage
       * collectable as soon as they are no longer referenced by any reader thread that may be in
       * the midst of traversing table right now.
       */

      final HashEntry<K, V>[] newTable = newEntryArray(oldCapacity << 1);
      threshold = (int) (newTable.length * loadFactor);
      final int sizeMask = newTable.length - 1;
      for (int i = 0; i < oldCapacity; i++) {
        // We need to guarantee that any existing reads of old Map can
        // proceed. So we cannot yet null out each bin.
        final HashEntry<K, V> e = oldTable[i];

        if (e != null) {
          final HashEntry<K, V> next = e.getNextEntry();
          final int idx = e.getEntryHash() & sizeMask;

          // Single node on list
          if (next == null) {
            newTable[idx] = e;
          } else {
            // Reuse trailing consecutive sequence at same slot
            HashEntry<K, V> lastRun = e;
            int lastIdx = idx;
            for (HashEntry<K, V> last = next; last != null; last = last.getNextEntry()) {
              final int k = last.getEntryHash() & sizeMask;
              if (k != lastIdx) {
                lastIdx = k;
                lastRun = last;
              }
            }
            newTable[lastIdx] = lastRun;

            // Clone all remaining nodes
            // Geode changes BEGIN
            // update the next entry instead of cloning the nodes in newTable;
            // this is primarily because we don't want to change
            // the underlying RegionEntry that may be used elsewhere;
            // however we create new wrapper entries for old table so that
            // iterators can continue on old table without blocking updates
            // for indefinite periods
            HashEntryImpl<K, V> newe, newp = null, newFirst = null;
            HashEntry<K, V> nextp;
            // Bug 44155 - we need to clone all of the entries, not just
            // the entries leading up to lastRun, because the entries
            // in the last run may have their next pointers changed
            // by a later rehash.
