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

import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX;

import java.io.File;
import java.io.IOException;
import java.io.Reader;
import java.lang.reflect.Array;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.SortedSet;
import java.util.StringTokenizer;
import java.util.TreeSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import io.micrometer.core.instrument.MeterRegistry;
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.GemFireConfigException;
import org.apache.geode.GemFireIOException;
import org.apache.geode.LogWriter;
import org.apache.geode.StatisticDescriptor;
import org.apache.geode.Statistics;
import org.apache.geode.StatisticsType;
import org.apache.geode.SystemConnectException;
import org.apache.geode.SystemFailure;
import org.apache.geode.alerting.internal.AlertingSession;
import org.apache.geode.alerting.internal.ClusterAlertMessaging;
import org.apache.geode.alerting.internal.InternalAlertingService;
import org.apache.geode.alerting.internal.InternalAlertingServiceFactory;
import org.apache.geode.alerting.internal.NullAlertMessaging;
import org.apache.geode.alerting.internal.spi.AlertLevel;
import org.apache.geode.annotations.Immutable;
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.CacheClosedException;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.DurableClientAttributes;
import org.apache.geode.distributed.internal.locks.GrantorRequestProcessor;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.api.MembershipInformation;
import org.apache.geode.distributed.internal.membership.api.MembershipLocator;
import org.apache.geode.distributed.internal.membership.api.QuorumChecker;
import org.apache.geode.distributed.internal.tcpserver.HostAddress;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.InternalInstantiator;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.admin.remote.DistributionLocatorId;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalCacheBuilder;
import org.apache.geode.internal.cache.execute.InternalFunctionService;
import org.apache.geode.internal.cache.execute.metrics.FunctionStatsManager;
import org.apache.geode.internal.cache.tier.sockets.EncryptorImpl;
import org.apache.geode.internal.cache.xmlcache.CacheServerCreation;
import org.apache.geode.internal.config.ClusterConfigurationNotAvailableException;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.logging.LogWriterFactory;
import org.apache.geode.internal.net.SocketCreatorFactory;
import org.apache.geode.internal.offheap.MemoryAllocator;
import org.apache.geode.internal.offheap.OffHeapStorage;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.security.SecurityServiceFactory;
import org.apache.geode.internal.statistics.DummyStatisticsRegistry;
import org.apache.geode.internal.statistics.GemFireStatSampler;
import org.apache.geode.internal.statistics.StatisticsConfig;
import org.apache.geode.internal.statistics.StatisticsManager;
import org.apache.geode.internal.statistics.StatisticsManagerFactory;
import org.apache.geode.internal.statistics.StatisticsRegistry;
import org.apache.geode.internal.statistics.platform.LinuxProcFsStatistics;
import org.apache.geode.internal.tcp.ConnectionTable;
import org.apache.geode.logging.internal.LoggingSession;
import org.apache.geode.logging.internal.NullLoggingSession;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.executors.LoggingUncaughtExceptionHandler;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.logging.internal.spi.LogConfig;
import org.apache.geode.logging.internal.spi.LogConfigListener;
import org.apache.geode.logging.internal.spi.LogConfigSupplier;
import org.apache.geode.logging.internal.spi.LogFile;
import org.apache.geode.management.ManagementException;
import org.apache.geode.metrics.internal.MeterRegistrySupplier;
import org.apache.geode.metrics.internal.MetricsService;
import org.apache.geode.pdx.internal.TypeRegistry;
import org.apache.geode.security.GemFireSecurityException;
import org.apache.geode.security.PostProcessor;
import org.apache.geode.security.SecurityManager;

/**
 * The concrete implementation of {@link DistributedSystem} that provides internal-only
 * functionality.
 *
 * @since GemFire 3.0
 */
public class InternalDistributedSystem extends DistributedSystem
    implements LogConfigSupplier {

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

  private static final String DISABLE_MANAGEMENT_PROPERTY =
      GEMFIRE_PREFIX + "disableManagement";

  public static final String ALLOW_MULTIPLE_SYSTEMS_PROPERTY =
      GEMFIRE_PREFIX + "ALLOW_MULTIPLE_SYSTEMS";

  public static final String ALLOW_MEMORY_OVERCOMMIT =
      GEMFIRE_PREFIX + "Cache.ALLOW_MEMORY_OVERCOMMIT";
  public static final String AVOID_MEMORY_LOCK_WHEN_OVERCOMMIT =
      GEMFIRE_PREFIX + "Cache.AVOID_MEMORY_LOCK_WHEN_OVERCOMMIT";

  /**
   * If auto-reconnect is going on this will hold a reference to it
   */
  @MakeNotStatic
  public static volatile DistributedSystem systemAttemptingReconnect;

  @Immutable
  public static final CreationStackGenerator DEFAULT_CREATION_STACK_GENERATOR = config -> null;

  // the following is overridden from DistributedTestCase to fix #51058
  @MutableForTesting
  public static final AtomicReference<CreationStackGenerator> TEST_CREATION_STACK_GENERATOR =
      new AtomicReference<>(DEFAULT_CREATION_STACK_GENERATOR);

  /**
   * A value of Boolean.TRUE will identify a thread being used to execute
   * disconnectListeners. {@link #addDisconnectListener} will not throw ShutdownException if the
   * value is Boolean.TRUE.
   */
  private final ThreadLocal<Boolean> isDisconnectThread =
      ThreadLocal.withInitial(() -> Boolean.FALSE);

  private final StatisticsManager statisticsManager;
  private MetricsService metricsService;
  private final FunctionStatsManager functionStatsManager;
  /**
   * True if the user is allowed lock when memory resources appear to be overcommitted.
   */
  private final boolean allowMemoryLockWhenOvercommitted =
      Boolean.getBoolean(ALLOW_MEMORY_OVERCOMMIT);
  /**
   * True if memory lock is avoided when memory resources appear to be overcommitted.
   */
  private final boolean avoidMemoryLockWhenOvercommitted =
      Boolean.getBoolean(AVOID_MEMORY_LOCK_WHEN_OVERCOMMIT);

  /**
   * The distribution manager that is used to communicate with the distributed system.
   */
  protected DistributionManager dm;

  private final GrantorRequestProcessor.GrantorRequestContext grc;

  private final AtomicReference<ClusterAlertMessaging> clusterAlertMessaging =
      new AtomicReference<>();

  // captured in initialize() when starting so that we can hand it to new instance when restarting
  private MembershipLocator<InternalDistributedMember> membershipLocator;

  /**
   * If the experimental multiple-system feature is enabled, always create a new system.
   *
   * <p>
   * Otherwise, create a new InternalDistributedSystem with the given properties, or connect to an
   * existing one with the same properties.
   */
  public static InternalDistributedSystem connectInternal(
      Properties config,
      SecurityConfig securityConfig,
      MetricsService.Builder metricsSessionBuilder) {
    return connectInternal(config, securityConfig, metricsSessionBuilder, null);
  }

  /**
   * If the experimental multiple-system feature is enabled, always create a new system.
   *
   * <p>
   * Otherwise, create a new InternalDistributedSystem with the given properties, or connect to an
   * existing one with the same properties.
   */
  public static InternalDistributedSystem connectInternal(
      Properties config,
      SecurityConfig securityConfig,
      MetricsService.Builder metricsSessionBuilder,
      final MembershipLocator<InternalDistributedMember> locator) {
    if (config == null) {
      config = new Properties();
    }

    if (Boolean.getBoolean(ALLOW_MULTIPLE_SYSTEMS_PROPERTY)) {
      return new Builder(config, metricsSessionBuilder)
          .setSecurityConfig(securityConfig)
          .setLocator(locator)
          .build();
    }

    synchronized (existingSystemsLock) {
      if (ClusterDistributionManager.isDedicatedAdminVM()) {
        // For a dedicated admin VM, check to see if there is already
        // a connect that will suit our purposes.
        InternalDistributedSystem existingSystem =
            (InternalDistributedSystem) getConnection(config);
        if (existingSystem != null) {
          return existingSystem;
        }

      } else {
        boolean existingSystemDisconnecting = true;
        boolean isReconnecting = false;
        while (!existingSystems.isEmpty() && existingSystemDisconnecting && !isReconnecting) {
          Assert.assertTrue(existingSystems.size() == 1);

          InternalDistributedSystem existingSystem = existingSystems.get(0);
          existingSystemDisconnecting = existingSystem.isDisconnecting();
          // a reconnecting DS will block on GemFireCache.class and a ReconnectThread
          // holds that lock and invokes this method, so we break out of the loop
          // if we detect this condition
          isReconnecting = existingSystem.isReconnectingDS();
          if (existingSystemDisconnecting) {
            boolean interrupted = Thread.interrupted();
            try {
              // no notify for existingSystemsLock, just to release the sync
              existingSystemsLock.wait(50);
            } catch (InterruptedException ex) {
              interrupted = true;
            } finally {
              if (interrupted) {
                Thread.currentThread().interrupt();
              }
            }
          } else if (existingSystem.isConnected()) {
            existingSystem.validateSameProperties(config, existingSystem.isConnected());
            return existingSystem;
          } else {
            throw new AssertionError(
                "system should not have both disconnecting==false and isConnected==false");
          }
        }
      }

      // Make a new connection to the distributed system
      InternalDistributedSystem newSystem = new Builder(config, metricsSessionBuilder)
          .setSecurityConfig(securityConfig)
          .setLocator(locator)
          .build();
      addSystem(newSystem);
      return newSystem;
    }
  }

  public GrantorRequestProcessor.GrantorRequestContext getGrantorRequestContext() {
    return grc;
  }

  /**
   * Numeric id that identifies this node in a DistributedSystem
   */
  private long id;

  /**
   * The log writer used to log information messages
   */
  @Deprecated
  protected InternalLogWriter logWriter = null;

  /**
   * The log writer used to log security related messages
   */
  @Deprecated
  protected InternalLogWriter securityLogWriter = null;

  /**
   * Distributed System clock
   */
  private DSClock clock;

  /**
   * Time this system was created
   */
  private final long startTime;

  /**
   * Guards access to {@link #isConnected}
   */
  private final Object isConnectedMutex = new Object();

  /**
   * Is this <code>DistributedSystem</code> connected to a distributed system?
   * <p>
   * Concurrency: volatile for reads and protected by synchronization of {@link #isConnectedMutex}
   * for writes
   */
  protected volatile boolean isConnected;

  /**
   * Set to true if this distributed system is a singleton; it will always be the only member of the
   * system.
   */
  private boolean isLoner = false;

  /**
   * The sampler for this DistributedSystem.
   */
  private GemFireStatSampler sampler = null;

  /**
   * A set of listeners that are invoked when this connection to the distributed system is
   * disconnected
   */
  private final Set<DisconnectListener> disconnectListeners = new LinkedHashSet<>();

  /**
   * Set of listeners that are invoked whenever a connection is created to the distributed system
   */
  @MakeNotStatic
  private static final Set<ConnectListener> connectListeners = new LinkedHashSet<>();

  /**
   * auto-reconnect listeners
   */
  @MakeNotStatic
  private static final List<ReconnectListener> reconnectListeners = new ArrayList<>();

  /**
   * whether this DS is one created to reconnect to the distributed system after a
   * forced-disconnect. This state is cleared once reconnect is successful.
   */
  private boolean isReconnectingDS;

  /**
   * During a reconnect attempt this is used to perform quorum checks before allowing a location
   * service to be started up in this JVM. If quorum checks fail then we delay starting location
   * services until a live locator can be contacted.
   */
  private QuorumChecker quorumChecker;


  /**
   * Due to Bug 38407, be careful about moving this to another class.
   */
  public static final String SHUTDOWN_HOOK_NAME = "Distributed system shutdown hook";
  /**
   * A property to prevent shutdown hooks from being registered with the VM. This is regarding bug
   * 38407
   */
  public static final String DISABLE_SHUTDOWN_HOOK_PROPERTY =
      GEMFIRE_PREFIX + "disableShutdownHook";

  /**
   * A property to append to existing log-file instead of truncating it.
   */
  private static final String APPEND_TO_LOG_FILE = GEMFIRE_PREFIX + "append-log";

  //////////////////// Configuration Fields ////////////////////

  /**
   * The config object used to create this distributed system
   */
  private final DistributionConfig originalConfig;

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

  /**
   * The config object to which most configuration work is delegated
   */
  private DistributionConfig config;

  private volatile boolean shareSockets;

  /**
   * if this distributed system starts a locator, it is stored here
   */
  private InternalLocator startedLocator;

  private final List<ResourceEventsListener> resourceListeners = new CopyOnWriteArrayList<>();

  private final boolean disableManagement = Boolean.getBoolean(DISABLE_MANAGEMENT_PROPERTY);

  /**
   * Stack trace showing the creation of this instance of InternalDistributedSystem.
   */
  private final Throwable creationStack;

  private volatile SecurityService securityService;

  /**
   * Used at client side, indicates whether the 'delta-propagation' property is enabled on the DS
   * this client is connected to. This variable is used to decide whether to send delta bytes or
   * full value to the server for a delta-update operation.
   */
  private boolean deltaEnabledOnServer = true;

  private final AlertingSession alertingSession;
  private final InternalAlertingService alertingService;

  private final LoggingSession loggingSession;
  private final Set<LogConfigListener> logConfigListeners = new HashSet<>();

  public boolean isDeltaEnabledOnServer() {
    return deltaEnabledOnServer;
  }

  public void setDeltaEnabledOnServer(boolean deltaEnabledOnServer) {
    this.deltaEnabledOnServer = deltaEnabledOnServer;
  }

  public static void removeSystem(InternalDistributedSystem oldSystem) {
    DistributedSystem.removeSystem(oldSystem);
  }

  /**
   * Returns a connection to the distributed system that is suitable for administration. For
   * administration, we are not as strict when it comes to existing connections.
   *
   * @since GemFire 4.0
   */
  public static DistributedSystem connectForAdmin(Properties props) {
    return DistributedSystem.connectForAdmin(props);
  }

  /**
   * Returns a connected distributed system for this VM, or null if there is no connected
   * distributed system in this VM. This method synchronizes on the existingSystems collection.
   * <p>
   * author bruce
   *
   * @since GemFire 5.0
   */
  public static InternalDistributedSystem getConnectedInstance() {
    InternalDistributedSystem result = null;
    synchronized (existingSystemsLock) {
      if (!existingSystems.isEmpty()) {
        InternalDistributedSystem existingSystem = existingSystems.get(0);
        if (existingSystem.isConnected()) {
          result = existingSystem;
        }
      }
    }
    return result;
  }

  /**
   * Returns the current distributed system, if there is one. Note: this method is no longer unsafe
   * size existingSystems uses copy-on-write.
   * <p>
   * author bruce
   *
   * @since GemFire 5.0
   */
  public static InternalDistributedSystem unsafeGetConnectedInstance() {
    InternalDistributedSystem result = getAnyInstance();
    if (result != null) {
      if (!result.isConnected()) {
        result = null;
      }
    }
    return result;
  }

  /**
   * @return distribution stats, or null if there is no distributed system available
   */
  public static DMStats getDMStats() {
    InternalDistributedSystem sys = getAnyInstance();
    if (sys != null && sys.dm != null) {
      return sys.dm.getStats();
    }
    return null;
  }

  /**
   * @return a log writer, or null if there is no distributed system available
   */
  @Deprecated
  public static LogWriter getLogger() {
    InternalDistributedSystem sys = getAnyInstance();
    if (sys != null && sys.logWriter != null) {
      return sys.logWriter;
    }
    return null;
  }

  @Deprecated
  public static InternalLogWriter getStaticInternalLogWriter() {
    InternalDistributedSystem sys = getAnyInstance();
    if (sys != null) {
      return sys.logWriter;
    }
    return null;
  }

  @Deprecated
  public InternalLogWriter getInternalLogWriter() {
    return logWriter;
  }

  @Deprecated
  public InternalLogWriter getSecurityInternalLogWriter() {
    InternalDistributedSystem sys = getAnyInstance();
    if (sys != null) {
      return sys.securityLogWriter;
    }
    return null;
  }

  /**
   * reset the reconnectAttempt counter for a new go at reconnecting
   */
  private static void resetReconnectAttemptCounter() {
    reconnectAttemptCounter.set(0);
  }

  /**
   * Creates a new {@code InternalDistributedSystem} with the given configuration.
   *
   * <p>
   * See {@link #connect} for a list of exceptions that may be thrown.
   *
   * @param config the configuration for the connection
   * @param statisticsManagerFactory creates the statistics manager for this member
   */
  private InternalDistributedSystem(ConnectionConfig config,
      StatisticsManagerFactory statisticsManagerFactory,
      FunctionStatsManager.Factory functionStatsManagerFactory) {
    alertingSession = AlertingSession.create();
    alertingService = new InternalAlertingServiceFactory().create();
    LoggingUncaughtExceptionHandler
        .setFailureSetter(SystemFailure::setFailure);
    loggingSession = LoggingSession.create();
    originalConfig = config.distributionConfig();
    isReconnectingDS = config.isReconnecting();
    quorumChecker = config.quorumChecker();

    // throws IllegalStateEx
    ((DistributionConfigImpl) originalConfig).checkForDisallowedDefaults();
    shareSockets = originalConfig.getConserveSockets();
    startTime = System.currentTimeMillis();
    grc = new GrantorRequestProcessor.GrantorRequestContext(stopper);

    creationStack =
        TEST_CREATION_STACK_GENERATOR.get().generateCreationStack(originalConfig);

    statisticsManager =
        statisticsManagerFactory.create(originalConfig.getName(), startTime, statsDisabled);

    functionStatsManager = functionStatsManagerFactory.create(statsDisabled, statisticsManager,
        new MeterRegistrySupplier(() -> this));
  }

  public SecurityService getSecurityService() {
    return securityService;
  }

  public void setSecurityService(SecurityService securityService) {
    this.securityService = securityService;
  }

  /**
   * Registers a listener to the system
   *
   * @param listener listener to be added
   */
  public void addResourceListener(ResourceEventsListener listener) {
    resourceListeners.add(listener);
  }

  /**
   * Un-Registers a listener to the system
   *
   * @param listener listener to be removed
   */
  public void removeResourceListener(ResourceEventsListener listener) {
    resourceListeners.remove(listener);
  }

  /**
   * @return the listeners registered with the system
   */
  public List<ResourceEventsListener> getResourceListeners() {
    return resourceListeners;
  }

  /**
   * Handles a particular event associated with a resource
   *
   * @param event Resource event
   * @param resource resource on which event is generated
   */
  public void handleResourceEvent(ResourceEvent event, Object resource) {
    if (disableManagement) {
      return;
    }
    if (resourceListeners.size() == 0) {
      return;
    }
    notifyResourceEventListeners(event, resource);
  }

  /**
   * Returns true if system is a loner (for testing)
   */
  public boolean isLoner() {
    return isLoner;
  }

  private MemoryAllocator offHeapStore = null;

  public MemoryAllocator getOffHeapStore() {
    return offHeapStore;
  }

  /**
   * Initializes this connection to a distributed system with the current configuration state.
   */
  @VisibleForTesting
  void initialize(SecurityManager securityManager, PostProcessor postProcessor,
      MetricsService.Builder metricsServiceBuilder,
      final MembershipLocator<InternalDistributedMember> membershipLocatorArg,
      ClusterDistributionManagerConstructor clusterDistributionManagerConstructor) {

    if (originalConfig.getLocators().equals("")) {
      if (originalConfig.getMcastPort() != 0) {
        throw new GemFireConfigException("The " + LOCATORS + " attribute can not be empty when the "
            + MCAST_PORT + " attribute is non-zero.");
      } else {
        // no distribution
        isLoner = true;
      }
    }

    config = new RuntimeDistributionConfigImpl(this);

    securityService = SecurityServiceFactory.create(
        config.getSecurityProps(),
        securityManager, postProcessor);

    if (!isLoner) {
      attemptingToReconnect = (reconnectAttemptCounter.get() > 0);
    }
    try {
      SocketCreatorFactory.setDistributionConfig(config);

      boolean logBanner = !attemptingToReconnect;
      boolean logConfiguration = !attemptingToReconnect;
      loggingSession.createSession(this, logBanner, logConfiguration);

      // LOG: create LogWriterLogger(s) for backwards compatibility of getLogWriter and
      // getSecurityLogWriter
      if (logWriter == null) {
        logWriter =
            LogWriterFactory.createLogWriterLogger(config, false);
        logWriter.fine("LogWriter is created.");
      }

      // logWriter.info("Created log writer for IDS@"+System.identityHashCode(this));

      if (securityLogWriter == null) {
        // LOG: whole new LogWriterLogger instance for security
        securityLogWriter =
            LogWriterFactory.createLogWriterLogger(config, true);
        securityLogWriter.fine("SecurityLogWriter is created.");
      }

      loggingSession.startSession();

      clock = new DSClock(isLoner);

      if (attemptingToReconnect && logger.isDebugEnabled()) {
        logger.debug(
            "This thread is initializing a new DistributedSystem in order to reconnect to other "
                + "members");
      }
      // Note we need loners to load the license in case they are a
      // cache server and will need to enforce the member limit
      if (Boolean.getBoolean(InternalLocator.FORCE_LOCATOR_DM_TYPE)) {
        locatorDMTypeForced = true;
      }

      InternalDataSerializer.initializeSerializationFilter(config);

      // Initialize the Diffie-Hellman and public/private keys
      try {
        EncryptorImpl.initCertsMap(config.getSecurityProps());
        EncryptorImpl.initPrivateKey(config.getSecurityProps());
        EncryptorImpl.initDHKeys(config);
      } catch (Exception ex) {
        throw new GemFireSecurityException(
            "Problem in initializing keys for client authentication",
            ex);
      }

      final long offHeapMemorySize =
          OffHeapStorage.parseOffHeapMemorySize(getConfig().getOffHeapMemorySize());

      offHeapStore = OffHeapStorage.createOffHeapStorage(this, offHeapMemorySize, this);

      // Note: this can only happen on a linux system
      if (getConfig().getLockMemory()) {
        // This calculation is not exact, but seems fairly close. So far we have
        // not loaded much into the heap and the current RSS usage is already
        // included the available memory calculation.
        long avail = LinuxProcFsStatistics.getAvailableMemory(logger);
        long size = offHeapMemorySize + Runtime.getRuntime().totalMemory();
        lockMemory(avail, size);
      }

      try {
        startInitLocator(membershipLocatorArg);
      } catch (InterruptedException e) {
        throw new SystemConnectException("Startup has been interrupted", e);
      }

      synchronized (isConnectedMutex) {
        isConnected = true;
      }

      if (!isLoner) {
        try {
          dm = clusterDistributionManagerConstructor.create(this, membershipLocator);
          // fix bug #46324
          if (InternalLocator.hasLocator()) {
            InternalLocator internalLocator = InternalLocator.getLocator();
            getDistributionManager().addHostedLocators(getDistributedMember(),
                InternalLocator.getLocatorStrings(),
                internalLocator.isSharedConfigurationEnabled());
          }
        } finally {
          if (dm == null && quorumChecker != null) {
            quorumChecker.resume();
          }
          setDisconnected();
        }
      } else {
        dm = new LonerDistributionManager(this, logWriter);
      }

      Assert.assertTrue(dm != null);
      Assert.assertTrue(dm.getSystem() == this);

      if (dm != null && dm.getDistribution() != null) {
        id = dm.getDistribution().getLocalMember().getMembershipPort();
      }

      synchronized (isConnectedMutex) {
        isConnected = true;
      }
      if (attemptingToReconnect && (startedLocator == null)) {
        try {
          startInitLocator(membershipLocatorArg);
        } catch (InterruptedException e) {
          throw new SystemConnectException("Startup has been interrupted", e);
        }
      }
      try {
        endInitLocator();
      } catch (IOException e) {
        throw new GemFireIOException("Problem finishing a locator service start", e);
      }

      startSampler();

      clusterAlertMessaging.set(new ClusterAlertMessaging(this));
      alertingService.useAlertMessaging(clusterAlertMessaging.get());
      alertingSession.createSession(alertingService);
      alertingSession.startSession();

      metricsService = metricsServiceBuilder.build(this);
      metricsService.start();

      // Log any instantiators that were registered before the log writer
      // was created
      InternalInstantiator.logInstantiators();
    } catch (RuntimeException ex) {
      if (startedLocator != null) {
        stopLocator();
      }
      config.close();
      throw ex;
    }

    reconnected = attemptingToReconnect;
    attemptingToReconnect = false;
    reconnectAttemptCounter.set(0);
  }

  void lockMemory(long avail, long size) {
    if (avail < size) {
      if (avoidMemoryLockWhenOvercommitted) {
        logger.warn(
            "System memory appears to be over committed by {} bytes. Memory will not be locked because -D{} is set to true.",
            size - avail, AVOID_MEMORY_LOCK_WHEN_OVERCOMMIT);
      } else if (allowMemoryLockWhenOvercommitted) {
        logger.warn(
            "System memory appears to be over committed by {} bytes. Memory is locked anyway because -D{} is set to true. You may experience instability, performance issues, or terminated processes due to the Linux OOM killer.",
            size - avail, ALLOW_MEMORY_OVERCOMMIT);
        lockMemory();
      } else {
        throw new IllegalStateException(
            String.format(
                "Insufficient free memory (%s) when attempting to lock %s bytes.  Either reduce the amount of heap or off-heap memory requested or free up additional system memory.  You may also specify -D%s=true on the command-line to override the constraint check.",
                avail, size, ALLOW_MEMORY_OVERCOMMIT));
      }
    } else {
      lockMemory();
    }
  }

  void lockMemory() {
    logger.info("Locking memory. This may take a while...");
    GemFireCacheImpl.lockMemory();
    logger.info("Finished locking memory.");
  }

  private void startSampler() {
    if (statsDisabled) {
      return;
    }
    sampler = loggingSession.getLogFile()
        .map(logFile -> new GemFireStatSampler(this, logFile))
        .orElseGet(() -> new GemFireStatSampler(this));
    sampler.start();
  }

  /**
   * Starts a locator in this JVM iff the distribution config wants one started.
   *
   * @since GemFire 5.7
   * @param membershipLocatorArg on initial startup, a MembershipLocator provided explicitly by
   *        a caller, or null; on restart, the old MembershipLocator (from the previous instance of
   *        InternalDistributedSystem.)
   */
  private void startInitLocator(
      final MembershipLocator<InternalDistributedMember> membershipLocatorArg)
      throws InterruptedException {

    final String locatorString = originalConfig.getStartLocator();
    if (locatorString.isEmpty()) {
      membershipLocator = membershipLocatorArg;
      return;
    }

    // when reconnecting we don't want to join with a colocated locator unless
    // there is a quorum of the old members available
    if (attemptingToReconnect && !isConnected) {
      if (quorumChecker != null) {
        logger.info("performing a quorum check to see if location services can be started early");
        if (!quorumChecker.checkForQuorum(3L * config.getMemberTimeout())) {
          logger.info("quorum check failed - not allowing location services to start early");
          return;
        }
        logger.info("Quorum check passed - allowing location services to start early");
      }
    }
    DistributionLocatorId locId = DistributionLocatorId.unmarshal(locatorString);
    try {
      startedLocator =
          InternalLocator.createLocator(locId.getPort(), NullLoggingSession.create(), null,
              logWriter, securityLogWriter, new HostAddress(locId.getHost()),
              locId.getHostnameForClients(), originalConfig.toProperties(),
              false);

      // if locator is started this way, cluster config is not enabled, set the flag correctly
