| /* |
| * 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.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Set; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.cache.control.ResourceManager; |
| 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.DistributionManager; |
| import org.apache.geode.distributed.internal.HighPriorityDistributionMessage; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.UpdateAttributesProcessor; |
| import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType; |
| import org.apache.geode.internal.cache.control.MemoryThresholds.MemoryState; |
| import org.apache.geode.internal.serialization.DeserializationContext; |
| import org.apache.geode.internal.serialization.SerializationContext; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| |
| /** |
| * The advisor associated with a {@link ResourceManager}. Allows knowledge of remote |
| * {@link ResourceManager} state and distribution of local {@link ResourceManager} state. |
| * |
| * @since GemFire 6.0 |
| */ |
| public class ResourceAdvisor extends DistributionAdvisor { |
| private static final Logger logger = LogService.getLogger(); |
| |
| /** |
| * Message used to push event updates to remote VMs |
| */ |
| public static class ResourceProfileMessage extends HighPriorityDistributionMessage { |
| // As of 9.0 this message will only ever have a single profile. |
| // But to be compatible with previous releases we still support |
| // multiple profiles so that we can handle these messages during |
| // a rolling upgrade. |
| private volatile ResourceManagerProfile[] profiles; |
| private volatile int processorId; |
| |
| /** |
| * Default constructor used for de-serialization (used during receipt) |
| */ |
| public ResourceProfileMessage() {} |
| |
| /** |
| * Constructor used to send profiles to other members. |
| * |
| * @param recips Members to send the profile to. |
| * @param profile Profile to send. |
| */ |
| private ResourceProfileMessage(final Set<InternalDistributedMember> recips, |
| final ResourceManagerProfile profile) { |
| setRecipients(recips); |
| this.processorId = 0; |
| this.profiles = new ResourceManagerProfile[] {profile}; |
| } |
| |
| @Override |
| protected void process(ClusterDistributionManager dm) { |
| Throwable thr = null; |
| ResourceManagerProfile p = null; |
| try { |
| final InternalCache cache = dm.getCache(); |
| if (cache != null && !cache.isClosed()) { |
| final ResourceAdvisor ra = cache.getInternalResourceManager().getResourceAdvisor(); |
| if (this.profiles != null) { |
| // Early reply to avoid waiting for the following putProfile call |
| // to fire (remote) listeners so that the origin member can proceed with |
| // firing its (local) listeners |
| |
| for (int i = 0; i < this.profiles.length; i++) { |
| ra.putProfile(this.profiles[i]); |
| } |
| } |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug("No cache: {}", this); |
| } |
| } |
| } catch (CancelException ignore) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Cache closed: {}", this); |
| } |
| } catch (VirtualMachineError err) { |
| SystemFailure.initiateFailure(err); |
| // If this ever returns, rethrow the error. We're poisoned |
| // now, so don't let this thread continue. |
| throw err; |
| } catch (Throwable t) { |
| // Whenever you catch Error or Throwable, you must also |
| // catch VirtualMachineError (see above). However, there is |
| // _still_ a possibility that you are dealing with a cascading |
| // error condition, so you also need to check to see if the JVM |
| // is still usable: |
| SystemFailure.checkFailure(); |
| thr = t; |
| } finally { |
| if (thr != null) { |
| dm.getCancelCriterion().checkCancelInProgress(null); |
| logger.info(String.format("This member caught exception processing profile %s %s", |
| new Object[] {p, toString()}), |
| thr); |
| } |
| } |
| } |
| |
| @Override |
| public int getDSFID() { |
| return RESOURCE_PROFILE_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| this.processorId = in.readInt(); |
| final int l = in.readInt(); |
| if (l != -1) { |
| this.profiles = new ResourceManagerProfile[l]; |
| for (int i = 0; i < this.profiles.length; i++) { |
| final ResourceManagerProfile r = new ResourceManagerProfile(); |
| InternalDataSerializer.invokeFromData(r, in); |
| this.profiles[i] = r; |
| } |
| } else { |
| this.profiles = null; |
| } |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| out.writeInt(this.processorId); |
| if (this.profiles != null) { |
| out.writeInt(this.profiles.length); |
| for (int i = 0; i < this.profiles.length; i++) { |
| InternalDataSerializer.invokeToData(this.profiles[i], out); |
| } |
| } else { |
| out.writeInt(-1); |
| } |
| } |
| |
| /** |
| * Send profiles to the provided members |
| * |
| * @param irm The resource manager which is requesting distribution |
| * @param recips The recipients of the message |
| * @param profile Profile to send in this message |
| */ |
| public static void send(final InternalResourceManager irm, |
| Set<InternalDistributedMember> recips, ResourceManagerProfile profile) { |
| final DistributionManager dm = irm.getResourceAdvisor().getDistributionManager(); |
| ResourceProfileMessage r = new ResourceProfileMessage(recips, profile); |
| dm.putOutgoing(r); |
| } |
| |
| @Override |
| public String getShortClassName() { |
| return "ResourceProfileMessage"; |
| } |
| |
| @Override |
| public String toString() { |
| StringBuilder sb = new StringBuilder(); |
| sb.append(getShortClassName()).append(" (processorId=").append(this.processorId) |
| .append("; profiles=["); |
| for (int i = 0; i < this.profiles.length; i++) { |
| sb.append(this.profiles[i]); |
| if (i < this.profiles.length - 1) { |
| sb.append(", "); |
| } |
| } |
| sb.append("]"); |
| sb.append(")"); |
| return sb.toString(); |
| } |
| } |
| |
| /** |
| * @param advisee The owner of this advisor |
| * @see ResourceManager |
| */ |
| private ResourceAdvisor(DistributionAdvisee advisee) { |
| super(advisee); |
| } |
| |
| public static ResourceAdvisor createResourceAdvisor(DistributionAdvisee advisee) { |
| ResourceAdvisor advisor = new ResourceAdvisor(advisee); |
| advisor.initialize(); |
| return advisor; |
| } |
| |
| @Override |
| protected Profile instantiateProfile(InternalDistributedMember memberId, int version) { |
| return new ResourceManagerProfile(memberId, version); |
| } |
| |
| private InternalResourceManager getResourceManager() { |
| return ((InternalCache) getAdvisee()).getInternalResourceManager(false); |
| } |
| |
| @SuppressWarnings("synthetic-access") |
| @Override |
| protected boolean evaluateProfiles(final Profile newProfile, final Profile oldProfile) { |
| |
| ResourceManagerProfile oldRMProfile = (ResourceManagerProfile) oldProfile; |
| ResourceManagerProfile newRMProfile = (ResourceManagerProfile) newProfile; |
| |
| List<ResourceEvent> eventsToDeliver = new ArrayList<ResourceEvent>(); |
| |
| if (oldRMProfile == null) { |
| eventsToDeliver.add(new MemoryEvent(ResourceType.HEAP_MEMORY, MemoryState.DISABLED, |
| newRMProfile.heapState, newRMProfile.getDistributedMember(), newRMProfile.heapBytesUsed, |
| false, newRMProfile.heapThresholds)); |
| eventsToDeliver.add(new MemoryEvent(ResourceType.OFFHEAP_MEMORY, MemoryState.DISABLED, |
| newRMProfile.offHeapState, newRMProfile.getDistributedMember(), |
| newRMProfile.offHeapBytesUsed, false, newRMProfile.offHeapThresholds)); |
| |
| } else { |
| if (oldRMProfile.heapState != newRMProfile.heapState) { |
| eventsToDeliver.add(new MemoryEvent(ResourceType.HEAP_MEMORY, oldRMProfile.heapState, |
| newRMProfile.heapState, newRMProfile.getDistributedMember(), newRMProfile.heapBytesUsed, |
| false, newRMProfile.heapThresholds)); |
| } |
| |
| if (newRMProfile.heapState == MemoryState.DISABLED) { |
| newRMProfile.setHeapData(oldRMProfile.heapBytesUsed, oldRMProfile.heapState, |
| oldRMProfile.heapThresholds); |
| } |
| |
| if (oldRMProfile.offHeapState != newRMProfile.offHeapState) { |
| eventsToDeliver.add(new MemoryEvent(ResourceType.OFFHEAP_MEMORY, oldRMProfile.offHeapState, |
| newRMProfile.offHeapState, newRMProfile.getDistributedMember(), |
| newRMProfile.offHeapBytesUsed, false, newRMProfile.offHeapThresholds)); |
| } |
| |
| if (newRMProfile.offHeapState == MemoryState.DISABLED) { |
| newRMProfile.setOffHeapData(oldRMProfile.offHeapBytesUsed, oldRMProfile.offHeapState, |
| oldRMProfile.offHeapThresholds); |
| } |
| } |
| |
| for (ResourceEvent event : eventsToDeliver) { |
| getResourceManager().deliverEventFromRemote(event); |
| } |
| |
| return true; |
| } |
| |
| @Override |
| public String toString() { |
| return new StringBuilder().append("ResourceAdvisor for ResourceManager ").append(getAdvisee()) |
| .toString(); |
| } |
| |
| /** |
| * Profile which shares state with other ResourceManagers. The data available in this profile |
| * should be enough to deliver a {@link MemoryEvent} for any of the CRITICAL {@link MemoryState}s |
| * |
| * @since GemFire 6.0 |
| */ |
| public static class ResourceManagerProfile extends Profile { |
| // Resource manager related fields |
| private long heapBytesUsed; |
| private MemoryState heapState; |
| private MemoryThresholds heapThresholds; |
| |
| private long offHeapBytesUsed; |
| private MemoryState offHeapState; |
| private MemoryThresholds offHeapThresholds; |
| |
| // Constructor for de-serialization |
| public ResourceManagerProfile() {} |
| |
| // Constructor for sending purposes |
| public ResourceManagerProfile(InternalDistributedMember memberId, int version) { |
| super(memberId, version); |
| } |
| |
| public synchronized ResourceManagerProfile setHeapData(final long heapBytesUsed, |
| final MemoryState heapState, final MemoryThresholds heapThresholds) { |
| this.heapBytesUsed = heapBytesUsed; |
| this.heapState = heapState; |
| this.heapThresholds = heapThresholds; |
| return this; |
| } |
| |
| public synchronized ResourceManagerProfile setOffHeapData(final long offHeapBytesUsed, |
| final MemoryState offHeapState, final MemoryThresholds offHeapThresholds) { |
| this.offHeapBytesUsed = offHeapBytesUsed; |
| this.offHeapState = offHeapState; |
| this.offHeapThresholds = offHeapThresholds; |
| return this; |
| } |
| |
| public synchronized MemoryEvent createDisabledMemoryEvent(ResourceType resourceType) { |
| if (resourceType == ResourceType.HEAP_MEMORY) { |
| return new MemoryEvent(ResourceType.HEAP_MEMORY, this.heapState, MemoryState.DISABLED, |
| getDistributedMember(), this.heapBytesUsed, false, this.heapThresholds); |
| } |
| |
| return new MemoryEvent(ResourceType.OFFHEAP_MEMORY, this.offHeapState, MemoryState.DISABLED, |
| getDistributedMember(), this.offHeapBytesUsed, false, this.offHeapThresholds); |
| } |
| |
| /** |
| * Used to process incoming Resource Manager profiles. A reply is expected to contain a profile |
| * with state of the local Resource Manager. |
| * |
| * @since GemFire 6.0 |
| */ |
| @Override |
| public void processIncoming(ClusterDistributionManager dm, String adviseePath, |
| boolean removeProfile, boolean exchangeProfiles, final List<Profile> replyProfiles) { |
| final InternalCache cache = dm.getCache(); |
| if (cache != null && !cache.isClosed()) { |
| handleDistributionAdvisee((DistributionAdvisee) cache, removeProfile, exchangeProfiles, |
| replyProfiles); |
| } |
| } |
| |
| @Override |
| public StringBuilder getToStringHeader() { |
| return new StringBuilder("ResourceAdvisor.ResourceManagerProfile"); |
| } |
| |
| @Override |
| public void fillInToString(StringBuilder sb) { |
| super.fillInToString(sb); |
| synchronized (this) { |
| sb.append("; heapState=").append(this.heapState).append("; heapBytesUsed=") |
| .append(this.heapBytesUsed).append("; heapThresholds=").append(this.heapThresholds) |
| .append("; offHeapState=").append(this.offHeapState).append("; offHeapBytesUsed=") |
| .append(this.offHeapBytesUsed).append("; offHeapThresholds=") |
| .append(this.offHeapThresholds); |
| } |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| |
| final long heapBytesUsed = in.readLong(); |
| MemoryState heapState = MemoryState.fromData(in); |
| MemoryThresholds heapThresholds = MemoryThresholds.fromData(in); |
| setHeapData(heapBytesUsed, heapState, heapThresholds); |
| |
| final long offHeapBytesUsed = in.readLong(); |
| MemoryState offHeapState = MemoryState.fromData(in); |
| MemoryThresholds offHeapThresholds = MemoryThresholds.fromData(in); |
| setOffHeapData(offHeapBytesUsed, offHeapState, offHeapThresholds); |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| final long heapBytesUsed; |
| final MemoryState heapState; |
| final MemoryThresholds heapThresholds; |
| final long offHeapBytesUsed; |
| final MemoryState offHeapState; |
| final MemoryThresholds offHeapThresholds; |
| synchronized (this) { |
| heapBytesUsed = this.heapBytesUsed; |
| heapState = this.heapState; |
| heapThresholds = this.heapThresholds; |
| |
| offHeapBytesUsed = this.offHeapBytesUsed; |
| offHeapState = this.offHeapState; |
| offHeapThresholds = this.offHeapThresholds; |
| } |
| super.toData(out, context); |
| |
| out.writeLong(heapBytesUsed); |
| heapState.toData(out); |
| heapThresholds.toData(out); |
| |
| out.writeLong(offHeapBytesUsed); |
| offHeapState.toData(out); |
| offHeapThresholds.toData(out); |
| } |
| |
| @Override |
| public int getDSFID() { |
| return RESOURCE_MANAGER_PROFILE; |
| } |
| |
| public synchronized MemoryState getHeapState() { |
| return this.heapState; |
| } |
| |
| public synchronized MemoryState getoffHeapState() { |
| return this.offHeapState; |
| } |
| } |
| |
| /** |
| * Get set of members whose {@linkplain ResourceManager#setCriticalHeapPercentage(float) critical |
| * heap threshold} has been met or exceeded. The set does not include the local VM. The mutability |
| * of this set only effects the elements in the set, not the state of the members. |
| * |
| * @return a mutable set of members in the critical state otherwise {@link Collections#EMPTY_SET} |
| */ |
| public Set<InternalDistributedMember> adviseCriticalMembers() { |
| return adviseFilter(new Filter() { |
| @Override |
| public boolean include(Profile profile) { |
| ResourceManagerProfile rmp = (ResourceManagerProfile) profile; |
| return rmp.getHeapState().isCritical(); |
| } |
| }); |
| } |
| |
| public boolean isHeapCritical(final InternalDistributedMember member) { |
| ResourceManagerProfile rmp = (ResourceManagerProfile) getProfile(member); |
| return rmp != null ? rmp.getHeapState().isCritical() : false; |
| } |
| |
| public synchronized void updateRemoteProfile() { |
| Set<InternalDistributedMember> recips = adviseGeneric(); |
| ResourceManagerProfile profile = |
| new ResourceManagerProfile(getDistributionManager().getId(), incrementAndGetVersion()); |
| getResourceManager().fillInProfile(profile); |
| ResourceProfileMessage.send(getResourceManager(), recips, profile); |
| } |
| |
| @Override |
| protected void profileRemoved(Profile profile) { |
| ResourceManagerProfile oldp = (ResourceManagerProfile) profile; |
| getResourceManager() |
| .deliverEventFromRemote(oldp.createDisabledMemoryEvent(ResourceType.HEAP_MEMORY)); |
| getResourceManager() |
| .deliverEventFromRemote(oldp.createDisabledMemoryEvent(ResourceType.OFFHEAP_MEMORY)); |
| } |
| |
| @Override |
| public void close() { |
| new UpdateAttributesProcessor(this.getAdvisee(), true).distribute(false); |
| super.close(); |
| } |
| } |
