geode-core/src/main/java/org/apache/geode/internal/cache/control/InternalResourceManager.java - geode - Git at Google

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


 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;

 import org.apache.logging.log4j.Logger;

 import org.apache.geode.CancelCriterion;
 import org.apache.geode.InternalGemFireError;
 import org.apache.geode.annotations.VisibleForTesting;
 import org.apache.geode.annotations.internal.MutableForTesting;
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.control.RebalanceFactory;
 import org.apache.geode.cache.control.RebalanceOperation;
 import org.apache.geode.cache.control.ResourceManager;
 import org.apache.geode.cache.control.RestoreRedundancyOperation;
 import org.apache.geode.distributed.DistributedMember;
 import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
 import org.apache.geode.distributed.internal.DistributionManager;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.cache.PartitionedRegion;
 import org.apache.geode.internal.cache.control.ResourceAdvisor.ResourceManagerProfile;
 import org.apache.geode.internal.cache.partitioned.LoadProbe;
 import org.apache.geode.internal.cache.partitioned.SizedBasedLoadProbe;
 import org.apache.geode.internal.classloader.ClassPathLoader;
 import org.apache.geode.internal.logging.CoreLoggingExecutors;
 import org.apache.geode.internal.monitoring.ThreadsMonitoring;
 import org.apache.geode.logging.internal.executors.LoggingExecutors;
 import org.apache.geode.logging.internal.log4j.api.LogService;
 import org.apache.geode.management.runtime.RestoreRedundancyResults;
 import org.apache.geode.util.internal.GeodeGlossary;

 /**
  * Implementation of ResourceManager with additional internal-only methods.
  */
 public class InternalResourceManager implements ResourceManager {
   private static final Logger logger = LogService.getLogger();

   final int MAX_RESOURCE_MANAGER_EXE_THREADS =
       Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "resource.manager.threads", 1);

   private final Collection<CompletableFuture<Void>> startupTasks = new ArrayList<>();

   public enum ResourceType {
     HEAP_MEMORY(0x1), OFFHEAP_MEMORY(0x2), MEMORY(0x3), ALL(0xFFFFFFFF);

     final int id;

     ResourceType(final int id) {
       this.id = id;
     }
   }

   private final Map<ResourceType, Set<ResourceListener<?>>> listeners = new HashMap<>();

   private final ScheduledExecutorService scheduledExecutor;
   private final ExecutorService notifyExecutor;

   // A map of in progress rebalance operations. The value is Boolean because ConcurrentHashMap does
   // not support null values.
   private final Map<RebalanceOperation, Boolean> inProgressRebalanceOperations =
       new ConcurrentHashMap<>();

   // A map of in progress restore redundancy completable futures. The value is Boolean because
   // ConcurrentHashMap does not support null values.
   private final Map<CompletableFuture<RestoreRedundancyResults>, Boolean> inProgressRedundancyOperations =
       new ConcurrentHashMap<>();

   final InternalCache cache;

   private LoadProbe loadProbe;

   private final ResourceManagerStats stats;
   private final ResourceAdvisor resourceAdvisor;
   private boolean closed;

   private final Map<ResourceType, ResourceMonitor> resourceMonitors;

   @MutableForTesting
   private static ResourceObserver observer = new ResourceObserverAdapter();

   private static final String PR_LOAD_PROBE_CLASS =
       System.getProperty(GeodeGlossary.GEMFIRE_PREFIX + "ResourceManager.PR_LOAD_PROBE_CLASS",
           SizedBasedLoadProbe.class.getName());

   public static InternalResourceManager getInternalResourceManager(Cache cache) {
     return (InternalResourceManager) cache.getResourceManager();
   }

   public static InternalResourceManager createResourceManager(final InternalCache cache) {
     return new InternalResourceManager(cache);
   }

   private InternalResourceManager(InternalCache cache) {
     this.cache = cache;
     resourceAdvisor = (ResourceAdvisor) cache.getDistributionAdvisor();
     stats = new ResourceManagerStats(cache.getDistributedSystem());

     // Create a new executor that other classes may use for handling resource
     // related tasks
     scheduledExecutor = LoggingExecutors
         .newScheduledThreadPool(MAX_RESOURCE_MANAGER_EXE_THREADS, "ResourceManagerRecoveryThread ");

     // Initialize the load probe
     try {
       Class<?> loadProbeClass = ClassPathLoader.getLatest().forName(PR_LOAD_PROBE_CLASS);
       loadProbe = (LoadProbe) loadProbeClass.newInstance();
     } catch (Exception e) {
       throw new InternalGemFireError("Unable to instantiate " + PR_LOAD_PROBE_CLASS, e);
     }

     // Create a new executor the resource manager and the monitors it creates
     // can use to handle dispatching of notifications.
     notifyExecutor =
         CoreLoggingExecutors.newSerialThreadPoolWithFeedStatistics(0,
             stats.getResourceEventQueueStatHelper(), "Notification Handler",
             thread -> thread.setPriority(Thread.MAX_PRIORITY), null,
             stats.getResourceEventPoolStatHelper(), getThreadMonitorObj());

     // Create the monitors
     Map<ResourceType, ResourceMonitor> tempMonitors = new HashMap<>();
     tempMonitors.put(ResourceType.HEAP_MEMORY,
         new HeapMemoryMonitor(this, cache, stats, new TenuredHeapConsumptionMonitor()));
     tempMonitors.put(ResourceType.OFFHEAP_MEMORY,
         new OffHeapMemoryMonitor(this, cache, cache.getOffHeapStore(), stats));
     resourceMonitors = Collections.unmodifiableMap(tempMonitors);

     // Initialize the listener sets so that it only needs to be done once
     for (ResourceType resourceType : new ResourceType[] {ResourceType.HEAP_MEMORY,
         ResourceType.OFFHEAP_MEMORY}) {
       Set<ResourceListener<?>> emptySet = new CopyOnWriteArraySet<>();
       listeners.put(resourceType, emptySet);
     }
   }

   public void close() {
     for (ResourceMonitor monitor : resourceMonitors.values()) {
       monitor.stopMonitoring();
     }

     stopExecutor(scheduledExecutor);
     stopExecutor(notifyExecutor);

     stats.close();
     closed = true;
   }

   boolean isClosed() {
     return closed;
   }

   public void fillInProfile(final Profile profile) {
     assert profile instanceof ResourceManagerProfile;

     for (ResourceMonitor monitor : resourceMonitors.values()) {
       monitor.fillInProfile((ResourceManagerProfile) profile);
     }
   }

   public void addResourceListener(final ResourceListener<?> listener) {
     addResourceListener(ResourceType.ALL, listener);
   }

   public void addResourceListener(final ResourceType resourceType,
       final ResourceListener<?> listener) {
     for (Map.Entry<ResourceType, Set<ResourceListener<?>>> mapEntry : listeners.entrySet()) {
       if ((mapEntry.getKey().id & resourceType.id) != 0) {
         mapEntry.getValue().add(listener);
       }
     }
   }

   public void removeResourceListener(final ResourceListener<?> listener) {
     removeResourceListener(ResourceType.ALL, listener);
   }

   public void removeResourceListener(final ResourceType resourceType,
       final ResourceListener<?> listener) {
     for (Map.Entry<ResourceType, Set<ResourceListener<?>>> mapEntry : listeners.entrySet()) {
       if ((mapEntry.getKey().id & resourceType.id) != 0) {
         mapEntry.getValue().remove(listener);
       }
     }
   }

   public Set<ResourceListener<?>> getResourceListeners(final ResourceType resourceType) {
     return listeners.get(resourceType);
   }

   /**
    * Deliver an event received from remote resource managers to the local listeners.
    *
    * @param event Event to deliver.
    */
   public void deliverEventFromRemote(final ResourceEvent event) {
     assert !event.isLocal();

     if (logger.isTraceEnabled()) {
       logger.trace(
           "New remote event to deliver for member " + event.getMember() + ": event=" + event);
     }

     runWithNotifyExecutor(() -> deliverLocalEvent(event));
   }

   void deliverLocalEvent(ResourceEvent event) {
     // Wait for an event to be handled by all listeners before starting to send another event
     synchronized (listeners) {
       resourceMonitors.get(event.getType()).notifyListeners(listeners.get(event.getType()), event);
     }

     stats.incResourceEventsDelivered();
   }

   public HeapMemoryMonitor getHeapMonitor() {
     return (HeapMemoryMonitor) resourceMonitors.get(ResourceType.HEAP_MEMORY);
   }

   public OffHeapMemoryMonitor getOffHeapMonitor() {
     return (OffHeapMemoryMonitor) resourceMonitors.get(ResourceType.OFFHEAP_MEMORY);
   }

   public MemoryMonitor getMemoryMonitor(boolean offheap) {
     if (offheap) {
       return getOffHeapMonitor();
     } else {
       return getHeapMonitor();
     }
   }

   /**
    * Use threshold event processor to execute the event embedded in the runnable.
    *
    * @param runnable Runnable to execute.
    */
   void runWithNotifyExecutor(Runnable runnable) {
     try {
       notifyExecutor.execute(runnable);
     } catch (RejectedExecutionException ignore) {
       if (!closed) {
         logger.warn("No memory events will be delivered because of RejectedExecutionException");
       }
     }
   }

   @Override
   public RebalanceFactory createRebalanceFactory() {
     return new RebalanceFactoryImpl();
   }

   @Override
   public Set<RebalanceOperation> getRebalanceOperations() {
     return Collections.unmodifiableSet(inProgressRebalanceOperations.keySet());
   }

   void addInProgressRebalance(RebalanceOperation op) {
     inProgressRebalanceOperations.put(op, Boolean.TRUE);
   }

   void removeInProgressRebalance(RebalanceOperation op) {
     inProgressRebalanceOperations.remove(op);
   }

   class RebalanceFactoryImpl implements RebalanceFactory {

     private Set<String> includedRegions;
     private Set<String> excludedRegions;

     @Override
     public RebalanceOperation simulate() {
       RegionFilter filter = new FilterByPath(includedRegions, excludedRegions);
       RebalanceOperationImpl op = new RebalanceOperationImpl(cache, true, filter);
       op.start();
       return op;
     }

     @Override
     public RebalanceOperation start() {
       RegionFilter filter = new FilterByPath(includedRegions, excludedRegions);
       RebalanceOperationImpl op = new RebalanceOperationImpl(cache, false, filter);
       op.start();
       return op;
     }

     @Override
     public RebalanceFactory excludeRegions(Set<String> regions) {
       excludedRegions = regions;
       return this;
     }

     @Override
     public RebalanceFactory includeRegions(Set<String> regions) {
       includedRegions = regions;
       return this;
     }
   }

   @Override
   public RestoreRedundancyOperation createRestoreRedundancyOperation() {
     return new RestoreRedundancyOperationImpl(cache);
   }

   @Override
   public Set<CompletableFuture<RestoreRedundancyResults>> getRestoreRedundancyFutures() {
     return Collections.unmodifiableSet(inProgressRedundancyOperations.keySet());
   }

   void addInProgressRestoreRedundancy(
       CompletableFuture<RestoreRedundancyResults> completableFuture) {
     inProgressRedundancyOperations.put(completableFuture, Boolean.TRUE);
   }

   void removeInProgressRestoreRedundancy(
       CompletableFuture<RestoreRedundancyResults> completableFuture) {
     inProgressRedundancyOperations.remove(completableFuture);
   }

   void stopExecutor(ExecutorService executor) {
     if (executor == null) {
       return;
     }
     executor.shutdown();
     final int secToWait =
         Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "prrecovery-close-timeout", 5);
     try {
       executor.awaitTermination(secToWait, TimeUnit.SECONDS);
     } catch (InterruptedException x) {
       Thread.currentThread().interrupt();
       logger.debug("Failed in interrupting the Resource Manager Thread due to interrupt");
     }
     if (!executor.isTerminated()) {
       logger.warn("Failed to stop resource manager threads in {} seconds, forcing shutdown",
           secToWait);
       List<Runnable> remainingTasks = executor.shutdownNow();
       remainingTasks.forEach(runnable -> {
         if (runnable instanceof Future) {
           ((Future<?>) runnable).cancel(true);
         }
       });
     }
   }

   public ScheduledExecutorService getExecutor() {
     return scheduledExecutor;
   }

   public ResourceManagerStats getStats() {
     return stats;
   }

   /**
    * For testing only, an observer which is called when rebalancing is started and finished for a
    * particular region. This observer is called even the "rebalancing" is actually redundancy
    * recovery for a particular region.
    */
   public static void setResourceObserver(ResourceObserver observer) {
     if (observer == null) {
       observer = new ResourceObserverAdapter();
     }
     InternalResourceManager.observer = observer;
   }


   /**
    * Get the resource observer used for testing. Never returns null.
    */
   public static ResourceObserver getResourceObserver() {
     return observer;
   }

   /**
    * For testing only. Receives callbacks for resource related events.
    */
   public interface ResourceObserver {
     /**
      * Indicates that rebalancing has started on a given region.
      */
     void rebalancingStarted(Region<?, ?> region);

     /**
      * Indicates that rebalancing has finished on a given region.
      */
     void rebalancingFinished(Region<?, ?> region);

     /**
      * Indicates that recovery has started on a given region.
      */
     void recoveryStarted(Region<?, ?> region);

     /**
      * Indicates that recovery has finished on a given region.
      */
     void recoveryFinished(Region<?, ?> region);

     /**
      * Indicated that a membership event triggered a recovery operation, but the recovery operation
      * will not be executed because there is already an existing recovery operation waiting to
      * happen on this region.
      */
     void recoveryConflated(PartitionedRegion region);

     /**
      * Indicates that a bucket is being moved from the source member to the target member.
      *
      * @param region the region
      * @param bucketId the bucket being moved
      * @param source the member the bucket is moving from
      * @param target the member the bucket is moving to
      */
     void movingBucket(Region<?, ?> region, int bucketId, DistributedMember source,
         DistributedMember target);

     /**
      * Indicates that a bucket primary is being moved from the source member to the target member.
      *
      * @param region the region
      * @param bucketId the bucket primary being moved
      * @param source the member the bucket primary is moving from
      * @param target the member the bucket primary is moving to
      */
     void movingPrimary(Region<?, ?> region, int bucketId, DistributedMember source,
         DistributedMember target);
   }

   public static class ResourceObserverAdapter implements ResourceObserver {

     @Override
     public void rebalancingFinished(Region<?, ?> region) {
       rebalancingOrRecoveryFinished(region);
     }

     @Override
     public void rebalancingStarted(Region<?, ?> region) {
       rebalancingOrRecoveryStarted(region);
     }

     @Override
     public void recoveryFinished(Region<?, ?> region) {
       rebalancingOrRecoveryFinished(region);
     }

     @Override
     public void recoveryStarted(Region<?, ?> region) {
       rebalancingOrRecoveryStarted(region);
     }

     /**
      * Indicated the a rebalance or a recovery has started.
      */
     public void rebalancingOrRecoveryStarted(Region<?, ?> region) {
       // do nothing
     }

     /**
      * Indicated the a rebalance or a recovery has finished.
      */
     public void rebalancingOrRecoveryFinished(Region<?, ?> region) {
       // do nothing
     }

     @Override
     public void recoveryConflated(PartitionedRegion region) {
       // do nothing
     }

     @Override
     public void movingBucket(Region<?, ?> region, int bucketId, DistributedMember source,
         DistributedMember target) {
       // do nothing
     }

     @Override
     public void movingPrimary(Region<?, ?> region, int bucketId, DistributedMember source,
         DistributedMember target) {
       // do nothing
     }
   }

   /*
    * (non-Javadoc)
    *
    * @see org.apache.geode.distributed.internal.DistributionAdvisee#getCancelCriterion()
    */
   public CancelCriterion getCancelCriterion() {
     return cache.getCancelCriterion();
   }

   public ResourceAdvisor getResourceAdvisor() {
     return resourceAdvisor;
   }

   public LoadProbe getLoadProbe() {
     return loadProbe;
   }

   /**
    * This method is test purposes only.
    */
   public LoadProbe setLoadProbe(LoadProbe probe) {
     LoadProbe old = loadProbe;
     loadProbe = probe;
     return old;
   }

   /*
    * (non-Javadoc)
    *
    * @see org.apache.geode.cache.control.ResourceManager#setEvictionHeapPercentage(int)
    */
   @Override
   public void setCriticalOffHeapPercentage(float offHeapPercentage) {
     getOffHeapMonitor().setCriticalThreshold(offHeapPercentage);
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public float getCriticalOffHeapPercentage() {
     return getOffHeapMonitor().getCriticalThreshold();
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public void setEvictionOffHeapPercentage(float offHeapPercentage) {
     getOffHeapMonitor().setEvictionThreshold(offHeapPercentage);
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public float getEvictionOffHeapPercentage() {
     return getOffHeapMonitor().getEvictionThreshold();
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public void setCriticalHeapPercentage(float heapPercentage) {
     getHeapMonitor().setCriticalThreshold(heapPercentage);
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public float getCriticalHeapPercentage() {
     return getHeapMonitor().getCriticalThreshold();
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public void setEvictionHeapPercentage(float heapPercentage) {
     getHeapMonitor().setEvictionThreshold(heapPercentage);
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public float getEvictionHeapPercentage() {
     return getHeapMonitor().getEvictionThreshold();
   }

   private ThreadsMonitoring getThreadMonitorObj() {
     DistributionManager distributionManager = cache.getDistributionManager();
     if (distributionManager != null) {
       return distributionManager.getThreadMonitoring();
     } else {
       return null;
     }
   }

   /**
    * Adds a task that represents an asynchronous action during startup
    *
    * @param startupTask the CompletableFuture startup task
    */
   public void addStartupTask(CompletableFuture<Void> startupTask) {
     Objects.requireNonNull(startupTask);
     synchronized (startupTasks) {
       startupTasks.add(startupTask);
     }
   }

   /**
    * Clears the startup tasks and returns a CompletableFuture that completes when all of the startup
    * tasks complete.
    *
    * @return a CompletableFuture that completes when all of the startup tasks complete
    */
   public CompletableFuture<Void> allOfStartupTasks() {
     synchronized (startupTasks) {
       CompletableFuture<?>[] completableFutures = startupTasks.toArray(new CompletableFuture[0]);
       startupTasks.clear();
       return CompletableFuture.allOf(completableFutures);
     }
   }

   @VisibleForTesting
   Collection<CompletableFuture<Void>> getStartupTasks() {
     return startupTasks;
   }
 }
	/*
	* 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.control;


	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.Objects;
	import java.util.Set;
	import java.util.concurrent.CompletableFuture;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.CopyOnWriteArraySet;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Future;
	import java.util.concurrent.RejectedExecutionException;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.TimeUnit;

	import org.apache.logging.log4j.Logger;

	import org.apache.geode.CancelCriterion;
	import org.apache.geode.InternalGemFireError;
	import org.apache.geode.annotations.VisibleForTesting;
	import org.apache.geode.annotations.internal.MutableForTesting;
	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.control.RebalanceFactory;
	import org.apache.geode.cache.control.RebalanceOperation;
	import org.apache.geode.cache.control.ResourceManager;
	import org.apache.geode.cache.control.RestoreRedundancyOperation;
	import org.apache.geode.distributed.DistributedMember;
	import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
	import org.apache.geode.distributed.internal.DistributionManager;
	import org.apache.geode.internal.cache.InternalCache;
	import org.apache.geode.internal.cache.PartitionedRegion;
	import org.apache.geode.internal.cache.control.ResourceAdvisor.ResourceManagerProfile;
	import org.apache.geode.internal.cache.partitioned.LoadProbe;
	import org.apache.geode.internal.cache.partitioned.SizedBasedLoadProbe;
	import org.apache.geode.internal.classloader.ClassPathLoader;
	import org.apache.geode.internal.logging.CoreLoggingExecutors;
	import org.apache.geode.internal.monitoring.ThreadsMonitoring;
	import org.apache.geode.logging.internal.executors.LoggingExecutors;
	import org.apache.geode.logging.internal.log4j.api.LogService;
	import org.apache.geode.management.runtime.RestoreRedundancyResults;
	import org.apache.geode.util.internal.GeodeGlossary;

	/**
	* Implementation of ResourceManager with additional internal-only methods.
	*/
	public class InternalResourceManager implements ResourceManager {
	private static final Logger logger = LogService.getLogger();

	final int MAX_RESOURCE_MANAGER_EXE_THREADS =
	Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "resource.manager.threads", 1);

	private final Collection<CompletableFuture<Void>> startupTasks = new ArrayList<>();

	public enum ResourceType {
	HEAP_MEMORY(0x1), OFFHEAP_MEMORY(0x2), MEMORY(0x3), ALL(0xFFFFFFFF);

	final int id;

	ResourceType(final int id) {
	this.id = id;
	}
	}

	private final Map<ResourceType, Set<ResourceListener<?>>> listeners = new HashMap<>();

	private final ScheduledExecutorService scheduledExecutor;
	private final ExecutorService notifyExecutor;

	// A map of in progress rebalance operations. The value is Boolean because ConcurrentHashMap does
	// not support null values.
	private final Map<RebalanceOperation, Boolean> inProgressRebalanceOperations =
	new ConcurrentHashMap<>();

	// A map of in progress restore redundancy completable futures. The value is Boolean because
	// ConcurrentHashMap does not support null values.
	private final Map<CompletableFuture<RestoreRedundancyResults>, Boolean> inProgressRedundancyOperations =
	new ConcurrentHashMap<>();

	final InternalCache cache;

	private LoadProbe loadProbe;

	private final ResourceManagerStats stats;
	private final ResourceAdvisor resourceAdvisor;
	private boolean closed;

	private final Map<ResourceType, ResourceMonitor> resourceMonitors;

	@MutableForTesting
	private static ResourceObserver observer = new ResourceObserverAdapter();

	private static final String PR_LOAD_PROBE_CLASS =
	System.getProperty(GeodeGlossary.GEMFIRE_PREFIX + "ResourceManager.PR_LOAD_PROBE_CLASS",
	SizedBasedLoadProbe.class.getName());

	public static InternalResourceManager getInternalResourceManager(Cache cache) {
	return (InternalResourceManager) cache.getResourceManager();
	}

	public static InternalResourceManager createResourceManager(final InternalCache cache) {
	return new InternalResourceManager(cache);
	}

	private InternalResourceManager(InternalCache cache) {
	this.cache = cache;
	resourceAdvisor = (ResourceAdvisor) cache.getDistributionAdvisor();
	stats = new ResourceManagerStats(cache.getDistributedSystem());

	// Create a new executor that other classes may use for handling resource
	// related tasks
	scheduledExecutor = LoggingExecutors
	.newScheduledThreadPool(MAX_RESOURCE_MANAGER_EXE_THREADS, "ResourceManagerRecoveryThread ");

	// Initialize the load probe
	try {
	Class<?> loadProbeClass = ClassPathLoader.getLatest().forName(PR_LOAD_PROBE_CLASS);
	loadProbe = (LoadProbe) loadProbeClass.newInstance();
	} catch (Exception e) {
	throw new InternalGemFireError("Unable to instantiate " + PR_LOAD_PROBE_CLASS, e);
	}

	// Create a new executor the resource manager and the monitors it creates
	// can use to handle dispatching of notifications.
	notifyExecutor =
	CoreLoggingExecutors.newSerialThreadPoolWithFeedStatistics(0,
	stats.getResourceEventQueueStatHelper(), "Notification Handler",
	thread -> thread.setPriority(Thread.MAX_PRIORITY), null,
	stats.getResourceEventPoolStatHelper(), getThreadMonitorObj());

	// Create the monitors
	Map<ResourceType, ResourceMonitor> tempMonitors = new HashMap<>();
	tempMonitors.put(ResourceType.HEAP_MEMORY,
	new HeapMemoryMonitor(this, cache, stats, new TenuredHeapConsumptionMonitor()));
	tempMonitors.put(ResourceType.OFFHEAP_MEMORY,
	new OffHeapMemoryMonitor(this, cache, cache.getOffHeapStore(), stats));
	resourceMonitors = Collections.unmodifiableMap(tempMonitors);

	// Initialize the listener sets so that it only needs to be done once
	for (ResourceType resourceType : new ResourceType[] {ResourceType.HEAP_MEMORY,
	ResourceType.OFFHEAP_MEMORY}) {
	Set<ResourceListener<?>> emptySet = new CopyOnWriteArraySet<>();
	listeners.put(resourceType, emptySet);
	}
	}

	public void close() {
	for (ResourceMonitor monitor : resourceMonitors.values()) {
	monitor.stopMonitoring();
	}

	stopExecutor(scheduledExecutor);
	stopExecutor(notifyExecutor);

	stats.close();
	closed = true;
	}

	boolean isClosed() {
	return closed;
	}

	public void fillInProfile(final Profile profile) {
	assert profile instanceof ResourceManagerProfile;

	for (ResourceMonitor monitor : resourceMonitors.values()) {
	monitor.fillInProfile((ResourceManagerProfile) profile);
	}
	}

	public void addResourceListener(final ResourceListener<?> listener) {
	addResourceListener(ResourceType.ALL, listener);
	}

	public void addResourceListener(final ResourceType resourceType,
	final ResourceListener<?> listener) {
	for (Map.Entry<ResourceType, Set<ResourceListener<?>>> mapEntry : listeners.entrySet()) {
	if ((mapEntry.getKey().id & resourceType.id) != 0) {
	mapEntry.getValue().add(listener);
	}
	}
	}

	public void removeResourceListener(final ResourceListener<?> listener) {
	removeResourceListener(ResourceType.ALL, listener);
	}

	public void removeResourceListener(final ResourceType resourceType,
	final ResourceListener<?> listener) {
	for (Map.Entry<ResourceType, Set<ResourceListener<?>>> mapEntry : listeners.entrySet()) {
	if ((mapEntry.getKey().id & resourceType.id) != 0) {
	mapEntry.getValue().remove(listener);
	}
	}
	}

	public Set<ResourceListener<?>> getResourceListeners(final ResourceType resourceType) {
	return listeners.get(resourceType);
	}

	/**
	* Deliver an event received from remote resource managers to the local listeners.
	*
	* @param event Event to deliver.
	*/
	public void deliverEventFromRemote(final ResourceEvent event) {
	assert !event.isLocal();

	if (logger.isTraceEnabled()) {
	logger.trace(
	"New remote event to deliver for member " + event.getMember() + ": event=" + event);
	}

	runWithNotifyExecutor(() -> deliverLocalEvent(event));
	}

	void deliverLocalEvent(ResourceEvent event) {
	// Wait for an event to be handled by all listeners before starting to send another event
	synchronized (listeners) {
	resourceMonitors.get(event.getType()).notifyListeners(listeners.get(event.getType()), event);
	}

	stats.incResourceEventsDelivered();
	}

	public HeapMemoryMonitor getHeapMonitor() {
	return (HeapMemoryMonitor) resourceMonitors.get(ResourceType.HEAP_MEMORY);
	}

	public OffHeapMemoryMonitor getOffHeapMonitor() {
	return (OffHeapMemoryMonitor) resourceMonitors.get(ResourceType.OFFHEAP_MEMORY);
	}

	public MemoryMonitor getMemoryMonitor(boolean offheap) {
	if (offheap) {
	return getOffHeapMonitor();
	} else {
	return getHeapMonitor();
	}
	}

	/**
	* Use threshold event processor to execute the event embedded in the runnable.
	*
	* @param runnable Runnable to execute.
	*/
	void runWithNotifyExecutor(Runnable runnable) {
	try {
	notifyExecutor.execute(runnable);
	} catch (RejectedExecutionException ignore) {
	if (!closed) {
	logger.warn("No memory events will be delivered because of RejectedExecutionException");
	}
	}
	}

	@Override
	public RebalanceFactory createRebalanceFactory() {
	return new RebalanceFactoryImpl();
	}

	@Override
	public Set<RebalanceOperation> getRebalanceOperations() {
	return Collections.unmodifiableSet(inProgressRebalanceOperations.keySet());
	}

	void addInProgressRebalance(RebalanceOperation op) {
	inProgressRebalanceOperations.put(op, Boolean.TRUE);
	}

	void removeInProgressRebalance(RebalanceOperation op) {
	inProgressRebalanceOperations.remove(op);
	}

	class RebalanceFactoryImpl implements RebalanceFactory {

	private Set<String> includedRegions;
	private Set<String> excludedRegions;

	@Override
	public RebalanceOperation simulate() {
	RegionFilter filter = new FilterByPath(includedRegions, excludedRegions);
	RebalanceOperationImpl op = new RebalanceOperationImpl(cache, true, filter);
	op.start();
	return op;
	}

	@Override
	public RebalanceOperation start() {
	RegionFilter filter = new FilterByPath(includedRegions, excludedRegions);
	RebalanceOperationImpl op = new RebalanceOperationImpl(cache, false, filter);
	op.start();
	return op;
	}

	@Override
	public RebalanceFactory excludeRegions(Set<String> regions) {
	excludedRegions = regions;
	return this;
	}

	@Override
	public RebalanceFactory includeRegions(Set<String> regions) {
	includedRegions = regions;
	return this;
	}
	}

	@Override
	public RestoreRedundancyOperation createRestoreRedundancyOperation() {
	return new RestoreRedundancyOperationImpl(cache);
	}

	@Override
	public Set<CompletableFuture<RestoreRedundancyResults>> getRestoreRedundancyFutures() {
	return Collections.unmodifiableSet(inProgressRedundancyOperations.keySet());
	}

	void addInProgressRestoreRedundancy(
	CompletableFuture<RestoreRedundancyResults> completableFuture) {
	inProgressRedundancyOperations.put(completableFuture, Boolean.TRUE);
	}

	void removeInProgressRestoreRedundancy(
	CompletableFuture<RestoreRedundancyResults> completableFuture) {
	inProgressRedundancyOperations.remove(completableFuture);
	}

	void stopExecutor(ExecutorService executor) {
	if (executor == null) {
	return;
	}
	executor.shutdown();
	final int secToWait =
	Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "prrecovery-close-timeout", 5);
	try {
	executor.awaitTermination(secToWait, TimeUnit.SECONDS);
	} catch (InterruptedException x) {
	Thread.currentThread().interrupt();
	logger.debug("Failed in interrupting the Resource Manager Thread due to interrupt");
	}
	if (!executor.isTerminated()) {
	logger.warn("Failed to stop resource manager threads in {} seconds, forcing shutdown",
	secToWait);
	List<Runnable> remainingTasks = executor.shutdownNow();
	remainingTasks.forEach(runnable -> {
	if (runnable instanceof Future) {
	((Future<?>) runnable).cancel(true);
	}
	});
	}
	}

	public ScheduledExecutorService getExecutor() {
	return scheduledExecutor;
	}

	public ResourceManagerStats getStats() {
	return stats;
	}

	/**
	* For testing only, an observer which is called when rebalancing is started and finished for a
	* particular region. This observer is called even the "rebalancing" is actually redundancy
	* recovery for a particular region.
	*/
	public static void setResourceObserver(ResourceObserver observer) {
	if (observer == null) {
	observer = new ResourceObserverAdapter();
	}
	InternalResourceManager.observer = observer;
	}


	/**
	* Get the resource observer used for testing. Never returns null.
	*/
	public static ResourceObserver getResourceObserver() {
	return observer;
	}

	/**
	* For testing only. Receives callbacks for resource related events.
	*/
	public interface ResourceObserver {
	/**
	* Indicates that rebalancing has started on a given region.
	*/
	void rebalancingStarted(Region<?, ?> region);

	/**
	* Indicates that rebalancing has finished on a given region.
	*/
	void rebalancingFinished(Region<?, ?> region);

	/**
	* Indicates that recovery has started on a given region.
	*/
	void recoveryStarted(Region<?, ?> region);

	/**
	* Indicates that recovery has finished on a given region.
	*/
	void recoveryFinished(Region<?, ?> region);

	/**
	* Indicated that a membership event triggered a recovery operation, but the recovery operation
	* will not be executed because there is already an existing recovery operation waiting to
	* happen on this region.
	*/
	void recoveryConflated(PartitionedRegion region);

	/**
	* Indicates that a bucket is being moved from the source member to the target member.
	*
	* @param region the region
	* @param bucketId the bucket being moved
	* @param source the member the bucket is moving from
	* @param target the member the bucket is moving to
	*/
	void movingBucket(Region<?, ?> region, int bucketId, DistributedMember source,
	DistributedMember target);

	/**
	* Indicates that a bucket primary is being moved from the source member to the target member.
	*
	* @param region the region
	* @param bucketId the bucket primary being moved
	* @param source the member the bucket primary is moving from
	* @param target the member the bucket primary is moving to
	*/
	void movingPrimary(Region<?, ?> region, int bucketId, DistributedMember source,
	DistributedMember target);
	}

	public static class ResourceObserverAdapter implements ResourceObserver {

	@Override
	public void rebalancingFinished(Region<?, ?> region) {
	rebalancingOrRecoveryFinished(region);
	}

	@Override
	public void rebalancingStarted(Region<?, ?> region) {
	rebalancingOrRecoveryStarted(region);
	}

	@Override
	public void recoveryFinished(Region<?, ?> region) {
	rebalancingOrRecoveryFinished(region);
	}

	@Override
	public void recoveryStarted(Region<?, ?> region) {
	rebalancingOrRecoveryStarted(region);
	}

	/**
	* Indicated the a rebalance or a recovery has started.
	*/
	public void rebalancingOrRecoveryStarted(Region<?, ?> region) {
	// do nothing
	}

	/**
	* Indicated the a rebalance or a recovery has finished.
	*/
	public void rebalancingOrRecoveryFinished(Region<?, ?> region) {
	// do nothing
	}

	@Override
	public void recoveryConflated(PartitionedRegion region) {
	// do nothing
	}

	@Override
	public void movingBucket(Region<?, ?> region, int bucketId, DistributedMember source,
	DistributedMember target) {
	// do nothing
	}

	@Override
	public void movingPrimary(Region<?, ?> region, int bucketId, DistributedMember source,
	DistributedMember target) {
	// do nothing
	}
	}

	/*
	* (non-Javadoc)
	*
	* @see org.apache.geode.distributed.internal.DistributionAdvisee#getCancelCriterion()
	*/
	public CancelCriterion getCancelCriterion() {
	return cache.getCancelCriterion();
	}

	public ResourceAdvisor getResourceAdvisor() {
	return resourceAdvisor;
	}

	public LoadProbe getLoadProbe() {
	return loadProbe;
	}

	/**
	* This method is test purposes only.
	*/
	public LoadProbe setLoadProbe(LoadProbe probe) {
	LoadProbe old = loadProbe;
	loadProbe = probe;
	return old;
	}

	/*
	* (non-Javadoc)
	*
	* @see org.apache.geode.cache.control.ResourceManager#setEvictionHeapPercentage(int)
	*/
	@Override
	public void setCriticalOffHeapPercentage(float offHeapPercentage) {
	getOffHeapMonitor().setCriticalThreshold(offHeapPercentage);
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public float getCriticalOffHeapPercentage() {
	return getOffHeapMonitor().getCriticalThreshold();
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public void setEvictionOffHeapPercentage(float offHeapPercentage) {
	getOffHeapMonitor().setEvictionThreshold(offHeapPercentage);
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public float getEvictionOffHeapPercentage() {
	return getOffHeapMonitor().getEvictionThreshold();
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public void setCriticalHeapPercentage(float heapPercentage) {
	getHeapMonitor().setCriticalThreshold(heapPercentage);
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public float getCriticalHeapPercentage() {
	return getHeapMonitor().getCriticalThreshold();
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public void setEvictionHeapPercentage(float heapPercentage) {
	getHeapMonitor().setEvictionThreshold(heapPercentage);
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public float getEvictionHeapPercentage() {
	return getHeapMonitor().getEvictionThreshold();
	}

	private ThreadsMonitoring getThreadMonitorObj() {
	DistributionManager distributionManager = cache.getDistributionManager();
	if (distributionManager != null) {
	return distributionManager.getThreadMonitoring();
	} else {
	return null;
	}
	}

	/**
	* Adds a task that represents an asynchronous action during startup
	*
	* @param startupTask the CompletableFuture startup task
	*/
	public void addStartupTask(CompletableFuture<Void> startupTask) {
	Objects.requireNonNull(startupTask);
	synchronized (startupTasks) {
	startupTasks.add(startupTask);
	}
	}

	/**
	* Clears the startup tasks and returns a CompletableFuture that completes when all of the startup
	* tasks complete.
	*
	* @return a CompletableFuture that completes when all of the startup tasks complete
	*/
	public CompletableFuture<Void> allOfStartupTasks() {
	synchronized (startupTasks) {
	CompletableFuture<?>[] completableFutures = startupTasks.toArray(new CompletableFuture[0]);
	startupTasks.clear();
	return CompletableFuture.allOf(completableFutures);
	}
	}

	@VisibleForTesting
	Collection<CompletableFuture<Void>> getStartupTasks() {
	return startupTasks;
	}
	}