/*
 * 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 java.lang.System.lineSeparator;
import static java.util.Collections.emptyList;
import static java.util.Collections.emptySet;
import static java.util.Collections.synchronizedMap;
import static java.util.Collections.unmodifiableCollection;
import static java.util.Collections.unmodifiableList;
import static java.util.Collections.unmodifiableMap;
import static java.util.Collections.unmodifiableSet;
import static java.util.Objects.requireNonNull;
import static java.util.concurrent.Executors.newScheduledThreadPool;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.commons.lang3.exception.ExceptionUtils.getStackTrace;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.cache.Region.SEPARATOR_CHAR;
import static org.apache.geode.distributed.ConfigurationPersistenceService.CLUSTER_CONFIG;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_MANAGER;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_POST_PROCESSOR;
import static org.apache.geode.distributed.internal.ClusterDistributionManager.ADMIN_ONLY_DM_TYPE;
import static org.apache.geode.distributed.internal.ClusterDistributionManager.LOCATOR_DM_TYPE;
import static org.apache.geode.distributed.internal.DistributionConfig.DEFAULT_DURABLE_CLIENT_ID;
import static org.apache.geode.distributed.internal.InternalDistributedSystem.getAnyInstance;
import static org.apache.geode.internal.cache.ColocationHelper.getColocatedChildRegions;
import static org.apache.geode.internal.cache.LocalRegion.setThreadInitLevelRequirement;
import static org.apache.geode.internal.cache.PartitionedRegion.DISK_STORE_FLUSHED;
import static org.apache.geode.internal.cache.PartitionedRegion.OFFLINE_EQUAL_PERSISTED;
import static org.apache.geode.internal.cache.PartitionedRegion.PRIMARY_BUCKETS_LOCKED;
import static org.apache.geode.internal.cache.PartitionedRegionHelper.PARTITION_LOCK_SERVICE_NAME;
import static org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType.HEAP_MEMORY;
import static org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType.OFFHEAP_MEMORY;
import static org.apache.geode.internal.logging.CoreLoggingExecutors.newThreadPoolWithFixedFeed;
import static org.apache.geode.internal.tcp.ConnectionTable.threadWantsSharedResources;
import static org.apache.geode.logging.internal.executors.LoggingExecutors.newFixedThreadPool;
import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.Reader;
import java.io.StringBufferInputStream;
import java.io.StringWriter;
import java.io.Writer;
import java.net.InetSocketAddress;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;

import javax.naming.Context;

import com.sun.jna.Native;
import com.sun.jna.Platform;
import io.micrometer.core.instrument.MeterRegistry;
import jakarta.transaction.TransactionManager;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.ForcedDisconnectException;
import org.apache.geode.GemFireCacheException;
import org.apache.geode.GemFireConfigException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.LogWriter;
import org.apache.geode.SerializationException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.admin.internal.SystemMemberCacheEventProcessor;
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.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheExistsException;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CacheXmlException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Declarable;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.DynamicRegionFactory;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.GatewayException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionService;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueFactoryImpl;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.asyncqueue.internal.InternalAsyncEventQueue;
import org.apache.geode.cache.client.ClientRegionFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.ClientMetadataService;
import org.apache.geode.cache.client.internal.ClientRegionFactoryImpl;
import org.apache.geode.cache.client.internal.InternalClientCache;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.control.ResourceManager;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.query.internal.InternalQueryService;
import org.apache.geode.cache.query.internal.QueryMonitor;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.CqServiceProvider;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.snapshot.CacheSnapshotService;
import org.apache.geode.cache.util.GatewayConflictResolver;
import org.apache.geode.cache.wan.GatewayReceiver;
import org.apache.geode.cache.wan.GatewayReceiverFactory;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.GatewaySenderFactory;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.Locator;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionAdvisee;
import org.apache.geode.distributed.internal.DistributionAdvisor;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.ResourceEvent;
import org.apache.geode.distributed.internal.ResourceEventsListener;
import org.apache.geode.distributed.internal.ServerLocationAndMemberId;
import org.apache.geode.distributed.internal.locks.DLockService;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.i18n.LogWriterI18n;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.backup.BackupService;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.ResourceAdvisor;
import org.apache.geode.internal.cache.event.EventTrackerExpiryTask;
import org.apache.geode.internal.cache.eviction.HeapEvictor;
import org.apache.geode.internal.cache.eviction.OffHeapEvictor;
import org.apache.geode.internal.cache.execute.util.FindRestEnabledServersFunction;
import org.apache.geode.internal.cache.extension.ExtensionPoint;
import org.apache.geode.internal.cache.extension.SimpleExtensionPoint;
import org.apache.geode.internal.cache.ha.HARegionQueue;
import org.apache.geode.internal.cache.locks.TXLockService;
import org.apache.geode.internal.cache.partitioned.RedundancyAlreadyMetException;
import org.apache.geode.internal.cache.partitioned.colocation.ColocationLoggerFactory;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager;
import org.apache.geode.internal.cache.snapshot.CacheSnapshotServiceImpl;
import org.apache.geode.internal.cache.tier.Acceptor;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
import org.apache.geode.internal.cache.tier.sockets.ClientHealthMonitor;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.internal.cache.wan.GatewaySenderAdvisor;
import org.apache.geode.internal.cache.wan.GatewaySenderQueueEntrySynchronizationListener;
import org.apache.geode.internal.cache.wan.InternalGatewaySender;
import org.apache.geode.internal.cache.wan.WANServiceProvider;
import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue;
import org.apache.geode.internal.cache.xmlcache.CacheServerCreation;
import org.apache.geode.internal.cache.xmlcache.CacheXmlGenerator;
import org.apache.geode.internal.cache.xmlcache.CacheXmlParser;
import org.apache.geode.internal.cache.xmlcache.CacheXmlPropertyResolver;
import org.apache.geode.internal.cache.xmlcache.PropertyResolver;
import org.apache.geode.internal.classloader.ClassPathLoader;
import org.apache.geode.internal.config.ClusterConfigurationNotAvailableException;
import org.apache.geode.internal.inet.LocalHostUtil;
import org.apache.geode.internal.jndi.JNDIInvoker;
import org.apache.geode.internal.jta.TransactionManagerImpl;
import org.apache.geode.internal.lang.ThrowableUtils;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
import org.apache.geode.internal.offheap.MemoryAllocator;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.security.SecurityServiceFactory;
import org.apache.geode.internal.sequencelog.SequenceLoggerImpl;
import org.apache.geode.internal.serialization.DSCODE;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.shared.StringPrintWriter;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.statistics.StatisticsClockFactory;
import org.apache.geode.internal.util.BlobHelper;
import org.apache.geode.internal.util.concurrent.FutureResult;
import org.apache.geode.lang.Identifiable;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.internal.JmxManagerAdvisee;
import org.apache.geode.management.internal.JmxManagerAdvisor;
import org.apache.geode.management.internal.RestAgent;
import org.apache.geode.management.internal.beans.ManagementListener;
import org.apache.geode.management.internal.configuration.domain.Configuration;
import org.apache.geode.management.internal.configuration.messages.ConfigurationResponse;
import org.apache.geode.pdx.JSONFormatter;
import org.apache.geode.pdx.PdxInstance;
import org.apache.geode.pdx.PdxInstanceFactory;
import org.apache.geode.pdx.PdxSerializer;
import org.apache.geode.pdx.ReflectionBasedAutoSerializer;
import org.apache.geode.pdx.internal.AutoSerializableManager;
import org.apache.geode.pdx.internal.InternalPdxInstance;
import org.apache.geode.pdx.internal.PdxInstanceFactoryImpl;
import org.apache.geode.pdx.internal.PdxInstanceImpl;
import org.apache.geode.pdx.internal.TypeRegistry;

/**
 * GemFire's implementation of a distributed {@link Cache}.
 */
public class GemFireCacheImpl implements InternalCache, InternalClientCache, HasCachePerfStats,
    DistributionAdvisee {
  private static final Logger logger = LogService.getLogger();

  /**
   * The default number of seconds to wait for a distributed lock
   */
  public static final int DEFAULT_LOCK_TIMEOUT =
      Integer.getInteger(GEMFIRE_PREFIX + "Cache.defaultLockTimeout", 60);

  /**
   * The default duration (in seconds) of a lease on a distributed lock
   */
  public static final int DEFAULT_LOCK_LEASE =
      Integer.getInteger(GEMFIRE_PREFIX + "Cache.defaultLockLease", 120);

  /**
   * The default "copy on read" attribute value
   */
  public static final boolean DEFAULT_COPY_ON_READ = false;

  /**
   * The default amount of time to wait for a {@code netSearch} to complete
   */
  public static final int DEFAULT_SEARCH_TIMEOUT =
      Integer.getInteger(GEMFIRE_PREFIX + "Cache.defaultSearchTimeout", 300);

  /**
   * Name of the default pool.
   */
  public static final String DEFAULT_POOL_NAME = "DEFAULT";

  @VisibleForTesting
  static final int EVENT_THREAD_LIMIT =
      Integer.getInteger(GEMFIRE_PREFIX + "Cache.EVENT_THREAD_LIMIT", 16);

  @VisibleForTesting
  static final int PURGE_INTERVAL = 1000;

  /**
   * The number of threads that the QueryMonitor will use to mark queries as cancelled (see
   * QueryMonitor class for reasons why a query might be cancelled). That processing is very
   * efficient, so we don't foresee needing to raise this above 1.
   */
  private static final int QUERY_MONITOR_THREAD_POOL_SIZE = 1;

  private static final int EVENT_QUEUE_LIMIT =
      Integer.getInteger(GEMFIRE_PREFIX + "Cache.EVENT_QUEUE_LIMIT", 4096);

  private static final int FIVE_HOURS_MILLIS = 5 * 60 * 60 * 1000;

  private static final Pattern DOUBLE_BACKSLASH = Pattern.compile("\\\\");

  /**
   * Enables {@link #creationStack} when CacheExistsException issues arise in debugging
   */
  private static final boolean DEBUG_CREATION_STACK = false;

  private static final boolean XML_PARAMETERIZATION_ENABLED =
      !Boolean.getBoolean(GEMFIRE_PREFIX + "xml.parameterization.disabled");

  /**
   * Number of threads used to close PRs in shutdownAll. By default is the number of PRs in the
   * cache.
   */
  private static final int shutdownAllPoolSize =
      Integer.getInteger(GEMFIRE_PREFIX + "SHUTDOWN_ALL_POOL_SIZE", -1);

  private static final ThreadLocal<GemFireCacheImpl> xmlCache = new ThreadLocal<>();

  private static final ThreadLocal<Thread> CLOSING_THREAD = new ThreadLocal<>();

  /**
   * System property to limit the max query-execution time. By default its turned off (-1), the time
   * is set in milliseconds.
   */
  @MutableForTesting
  public static int MAX_QUERY_EXECUTION_TIME =
      Integer.getInteger(GEMFIRE_PREFIX + "Cache.MAX_QUERY_EXECUTION_TIME", -1);

  /**
   * Used by unit tests to force cache creation to use a test generated cache.xml
   */
  @MutableForTesting
  @VisibleForTesting
  public static File testCacheXml;

  /**
   * If true then when a delta is applied the size of the entry value will be recalculated. If false
   * (the default) then the size of the entry value is unchanged by a delta application. Not a final
   * so that tests can change this value.
   *
   * @since GemFire h****** 6.1.2.9
   */
  @MutableForTesting
  static boolean DELTAS_RECALCULATE_SIZE =
      Boolean.getBoolean(GEMFIRE_PREFIX + "DELTAS_RECALCULATE_SIZE");

  /**
   * The {@code CacheLifecycleListener} s that have been registered in this VM
   */
  @MakeNotStatic
  private static final Set<CacheLifecycleListener> cacheLifecycleListeners =
      new CopyOnWriteArraySet<>();

  /**
   * TODO: remove static from defaultDiskStoreName and move methods to InternalCache
   */
  @MakeNotStatic
  private static String defaultDiskStoreName = DiskStoreFactory.DEFAULT_DISK_STORE_NAME;

  /**
   * System property to disable query monitor even if resource manager is in use
   */
  private final boolean queryMonitorDisabledForLowMem =
      Boolean.getBoolean(GEMFIRE_PREFIX + "Cache.DISABLE_QUERY_MONITOR_FOR_LOW_MEMORY");

  private final InternalDistributedSystem system;

  private final DistributionManager dm;

  private final ConcurrentMap<String, InternalRegion> rootRegions;

  /**
   * True if this cache is being created by a ClientCacheFactory.
   */
  private final boolean isClient;

  private final PoolFactory poolFactory;

  private final ConcurrentMap<String, InternalRegion> pathToRegion = new ConcurrentHashMap<>();

  private final CachePerfStats cachePerfStats;

  /**
   * Date on which this instances was created
   */
  private final Date creationDate;

  /**
   * Thread pool for event dispatching
   */
  private final ExecutorService eventThreadPool;

  /**
   * List of all cache servers. CopyOnWriteArrayList is used to allow concurrent add, remove and
   * retrieval operations. It is assumed that the traversal operations on cache servers list vastly
   * outnumber the mutative operations such as add, remove.
   */
  private final List<InternalCacheServer> allCacheServers = new CopyOnWriteArrayList<>();

  /**
   * Unmodifiable view of "allCacheServers".
   */
  private final List<CacheServer> unmodifiableAllCacheServers = unmodifiableList(allCacheServers);

  /**
   * Controls updates to the list of all gateway senders {@link #allGatewaySenders}.
   */
  private final Object allGatewaySendersLock = new Object();

  /**
   * List of all async event queues added to the cache. CopyOnWriteArrayList is used to allow
   * concurrent add, remove and retrieval operations.
   */
  private final Set<AsyncEventQueue> allVisibleAsyncEventQueues = new CopyOnWriteArraySet<>();

  /**
   * List of all async event queues added to the cache. CopyOnWriteArrayList is used to allow
   * concurrent add, remove and retrieval operations.
   */
  private final Set<AsyncEventQueue> allAsyncEventQueues = new CopyOnWriteArraySet<>();

  private final AtomicReference<GatewayReceiver> gatewayReceiver = new AtomicReference<>();

  private final AtomicReference<InternalCacheServer> gatewayReceiverServer =
      new AtomicReference<>();

  /**
   * PartitionedRegion instances (for required-events notification
   */
  private final Set<PartitionedRegion> partitionedRegions = new HashSet<>();

  /**
   * Map of regions that are in the process of being destroyed. We could potentially leave the
   * regions in the pathToRegion map, but that would entail too many changes at this point in the
   * release. We need to know which regions are being destroyed so that a profile exchange can get
   * the persistent id of the destroying region and know not to persist that ID if it receives it as
   * part of the persistent view.
   */
  private final ConcurrentMap<String, DistributedRegion> regionsInDestroy =
      new ConcurrentHashMap<>();

  private final Object allGatewayHubsLock = new Object();

  /**
   * Transaction manager for this cache.
   */
  private final TXManagerImpl transactionManager;

  /**
   * Named region attributes registered with this cache.
   */
  private final Map<String, RegionAttributes<?, ?>> namedRegionAttributes =
      synchronizedMap(new HashMap<>());

  /**
   * System timer task for cleaning up old bridge thread event entries.
   */
  private final EventTrackerExpiryTask recordedEventSweeper;

  private final TombstoneService tombstoneService;

  /**
   * Synchronization mutex for prLockService.
   */
  private final Object prLockServiceLock = new Object();

  /**
   * Synchronization mutex for gatewayLockService.
   */
  private final Object gatewayLockServiceLock = new Object();

  private final InternalResourceManager resourceManager;

  private final BackupService backupService;

  private final Object heapEvictorLock = new Object();

  private final Object offHeapEvictorLock = new Object();

  private final Object queryMonitorLock = new Object();

  private final PersistentMemberManager persistentMemberManager;

  private final ClientMetadataService clientMetadataService;

  private final AtomicBoolean isShutDownAll = new AtomicBoolean();
  private final CountDownLatch shutDownAllFinished = new CountDownLatch(1);

  private final ResourceAdvisor resourceAdvisor;
  private final JmxManagerAdvisor jmxAdvisor;

  private final int serialNumber;

  private final TXEntryStateFactory txEntryStateFactory;

  private final CacheConfig cacheConfig;

  private final DiskStoreMonitor diskMonitor;

  /**
   * Stores the properties used to initialize declarables.
   */
  private final Map<Declarable, Properties> declarablePropertiesMap = new ConcurrentHashMap<>();

  /**
   * Resolves ${} type property strings.
   */
  private final PropertyResolver resolver;

  /**
   * @since GemFire 8.1
   */
  private final ExtensionPoint<Cache> extensionPoint = new SimpleExtensionPoint<>(this, this);

  private final CqService cqService;

  private final Set<RegionListener> regionListeners = ConcurrentHashMap.newKeySet();

  private final Map<Class<? extends CacheService>, CacheService> services = new HashMap<>();

  private final SecurityService securityService;

  private final Set<RegionEntrySynchronizationListener> synchronizationListeners =
      ConcurrentHashMap.newKeySet();

  private final ClusterConfigurationLoader clusterConfigurationLoader =
      new ClusterConfigurationLoader();

  private final StatisticsClock statisticsClock;

  /**
   * Map of futures used to track regions that are being reinitialized.
   */
  private final ConcurrentMap<String, FutureResult<InternalRegion>> reinitializingRegions =
      new ConcurrentHashMap<>();

  private final HeapEvictorFactory heapEvictorFactory;
  private final Runnable typeRegistryClose;
  private final Function<InternalCache, String> typeRegistryGetPdxDiskStoreName;
  private final Consumer<PdxSerializer> typeRegistrySetPdxSerializer;
  private final TypeRegistryFactory typeRegistryFactory;
  private final Consumer<org.apache.geode.cache.execute.Function<?>> functionServiceRegisterFunction;
  private final Function<Object, SystemTimer> systemTimerFactory;
  private final ReplyProcessor21Factory replyProcessor21Factory;

  private final Stopper stopper = new Stopper();

  private final boolean disableDisconnectDsOnCacheClose =
      Boolean.getBoolean(GEMFIRE_PREFIX + "DISABLE_DISCONNECT_DS_ON_CACHE_CLOSE");

  private final ConcurrentMap<String, DiskStoreImpl> diskStores = new ConcurrentHashMap<>();

  private final ConcurrentMap<String, DiskStoreImpl> regionOwnedDiskStores =
      new ConcurrentHashMap<>();

  /**
   * Synchronization mutex for {@link #ccpTimer}.
   */
  private final Object ccpTimerMutex = new Object();

  private final ExpirationScheduler expirationScheduler;

  /**
   * TODO make this a simple int guarded by riWaiters and get rid of the double-check
   */
  private final AtomicInteger registerInterestsInProgress = new AtomicInteger();

  private final List<SimpleWaiter> riWaiters = new ArrayList<>();

  private final InternalCacheForClientAccess cacheForClients =
      new InternalCacheForClientAccess(this);

  private volatile ConfigurationResponse configurationResponse;

  private volatile boolean isInitialized;

  private volatile boolean isClosing;

  private final CountDownLatch isClosedLatch = new CountDownLatch(1);

  /**
   * Set of all gateway senders. It may be fetched safely (for enumeration), but updates must by
   * synchronized via {@link #allGatewaySendersLock}
   */
  private volatile Set<GatewaySender> allGatewaySenders = emptySet();

  /**
   * Copy on Read feature for all read operations.
   */
  private volatile boolean copyOnRead = DEFAULT_COPY_ON_READ;

  /**
   * Reason this cache was forced to close due to a forced-disconnect or system failure.
   */
  private volatile Throwable disconnectCause;

  /**
   * DistributedLockService for GatewaySenders. Remains null until the first GatewaySender is
   * created. Destroyed by GemFireCache when closing the cache. Guarded by gatewayLockServiceLock.
   */
  private volatile DistributedLockService gatewayLockService;

  private volatile QueryMonitor queryMonitor;

  /**
   * Not final to allow cache.xml parsing to set it.
   */
  private Pool defaultPool;

  /**
   * Amount of time (in seconds) to wait for a distributed lock
   */
  private int lockTimeout = DEFAULT_LOCK_TIMEOUT;

  /**
   * Amount of time a lease of a distributed lock lasts
   */
  private int lockLease = DEFAULT_LOCK_LEASE;

  /**
   * Amount of time to wait for a {@code netSearch} to complete
   */
  private int searchTimeout = DEFAULT_SEARCH_TIMEOUT;

  /**
   * Conflict resolver for WAN, if any. Guarded by {@link #allGatewayHubsLock}.
   */
  private GatewayConflictResolver gatewayConflictResolver;

  /**
   * True if this is a server cache.
   */
  private boolean isServer;

  private RestAgent restAgent;

  private boolean isRESTServiceRunning;

  /**
   * True if this cache was forced to close due to a forced-disconnect.
   */
  private boolean forcedDisconnect;

  /**
   * Context where this cache was created for debugging.
   */
  private Exception creationStack;

  /**
   * DistributedLockService for PartitionedRegions. Remains null until the first PartitionedRegion
   * is created. Destroyed by GemFireCache when closing the cache. Protected by synchronization on
   * prLockService. Guarded by prLockServiceLock.
   */
  private DistributedLockService prLockService;

  private HeapEvictor heapEvictor;

  private OffHeapEvictor offHeapEvictor;

  private ResourceEventsListener resourceEventsListener;

  /**
   * Set to true during a cache close if user requested durable subscriptions to be kept.
   *
   * @since GemFire 5.7
   */
  private boolean keepAlive;

  /**
   * Timer for {@link CacheClientProxy}. Guarded by {@link #ccpTimerMutex}.
   */
  private SystemTimer ccpTimer;

  private int cancelCount;

  /**
   * Some tests pass value in via constructor. Product code sets value after construction.
   */
  private TypeRegistry pdxRegistry;

  private Declarable initializer;

  private Properties initializerProps;

  private List<File> backupFiles = emptyList();

  private final ConcurrentMap<String, CountDownLatch> diskStoreLatches = new ConcurrentHashMap<>();

  static {
    // this works around jdk bug 6427854
    String propertyName = "sun.nio.ch.bugLevel";
    String value = System.getProperty(propertyName);
    if (value == null) {
      System.setProperty(propertyName, "");
    }
  }

  /**
   * Invokes mlockall(). Locks all pages mapped into the address space of the calling process. This
   * includes the pages of the code, data and stack segment, as well as shared libraries, user space
   * kernel data, shared memory, and memory-mapped files. All mapped pages are guaranteed to be
   * resident in RAM when the call returns successfully; the pages are guaranteed to stay in RAM
   * until later unlocked.
   *
   * @param flags MCL_CURRENT 1 - Lock all pages which are currently mapped into the address space
   *        of the process.
   *
   *        MCL_FUTURE 2 - Lock all pages which will become mapped into the address space of the
   *        process in the future. These could be for instance new pages required by a growing heap
   *        and stack as well as new memory mapped files or shared memory regions.
   *
   * @return 0 if success, non-zero if error and errno set
   */
  private static native int mlockall(int flags);

  public static void lockMemory() {
    try {
      Native.register(Platform.C_LIBRARY_NAME);
      int result = mlockall(1);
      if (result == 0) {
        return;
      }
    } catch (Throwable t) {
      throw new IllegalStateException("Error trying to lock memory", t);
    }

    int lastError = Native.getLastError();
    String message = "mlockall failed: " + lastError;
    if (lastError == 1 || lastError == 12) {
      // if EPERM || ENOMEM
      message = "Unable to lock memory due to insufficient free space or privileges.  "
          + "Please check the RLIMIT_MEMLOCK soft resource limit (ulimit -l) and "
          + "increase the available memory if needed";
    }
    throw new IllegalStateException(message);
  }

  /**
   * Returns the last created instance of GemFireCache.
   *
   * @deprecated use DM.getCache instead
   */
  @Deprecated
  public static GemFireCacheImpl getInstance() {
    InternalDistributedSystem system = getAnyInstance();
    if (system == null) {
      return null;
    }

    GemFireCacheImpl cache = (GemFireCacheImpl) system.getCache();
    if (cache == null) {
      return null;
    }

    if (cache.isClosing) {
      return null;
    }

    return cache;
  }

  /**
   * Returns an existing instance. If a cache does not exist throws a cache closed exception.
   *
   * @return the existing cache
   * @throws CacheClosedException if an existing cache can not be found.
   * @deprecated use DM.getExistingCache instead.
   */
  @Deprecated
  public static GemFireCacheImpl getExisting() {
    GemFireCacheImpl result = getInstance();
    if (result != null && !result.isClosing) {
      return result;
    }
    if (result != null) {
      throw result.getCacheClosedException("The cache has been closed.");
    }
    throw new CacheClosedException("A cache has not yet been created.");
  }

  /**
