You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ud...@apache.org on 2018/07/31 19:55:19 UTC
[geode] branch feature/CleanMicrometer updated: WIP: cleaned up
imports Add GatewayReceiverStats.kt GatewaySenderStats.kt
This is an automated email from the ASF dual-hosted git repository.
udo pushed a commit to branch feature/CleanMicrometer
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/feature/CleanMicrometer by this push:
new 7f2d221 WIP: cleaned up imports Add GatewayReceiverStats.kt GatewaySenderStats.kt
7f2d221 is described below
commit 7f2d221449d8bc809b11dc824b45b92a48b66706
Author: Udo Kohlmeyer <uk...@pivotal.io>
AuthorDate: Tue Jul 31 12:55:06 2018 -0700
WIP: cleaned up imports
Add GatewayReceiverStats.kt GatewaySenderStats.kt
---
.../org/apache/geode/cache/query/internal/Bag.java | 2 +-
.../cache/query/internal/ExecutionContext.java | 2 +-
.../geode/cache/query/internal/ResultsBag.java | 2 +-
.../internal/ClusterDistributionManager.java | 482 +++++++--------
.../distributed/internal/locks/DLockService.java | 375 ++++++------
.../internal/membership/MembershipManager.java | 2 +-
.../apache/geode/internal/cache/BucketAdvisor.java | 1 +
.../apache/geode/internal/cache/BucketRegion.java | 9 +-
.../geode/internal/cache/DistributedRegion.java | 8 +-
.../cache/InternalCacheForClientAccess.java | 1 +
.../geode/internal/cache/PartitionedRegion.java | 33 +-
.../internal/cache/PartitionedRegionDataStore.java | 13 +-
.../cache/PartitionedRegionRedundancyTracker.java | 3 +-
.../cache/partitioned/ContainsKeyValueMessage.java | 3 +-
.../internal/cache/partitioned/DumpB2NRegion.java | 4 +-
.../cache/partitioned/FetchBulkEntriesMessage.java | 3 +-
.../cache/partitioned/FetchEntriesMessage.java | 3 +-
.../cache/partitioned/FetchKeysMessage.java | 3 +-
.../internal/cache/partitioned/GetMessage.java | 67 +--
.../cache/partitioned/IdentityRequestMessage.java | 3 +-
.../cache/partitioned/InterestEventMessage.java | 3 +-
.../cache/partitioned/ManageBucketMessage.java | 3 +-
.../internal/cache/partitioned/RegionAdvisor.java | 2 +-
.../internal/cache/partitioned/SizeMessage.java | 3 +-
.../cache/persistence/PersistenceAdvisorImpl.java | 2 +-
.../internal/cache/tier/sockets/AcceptorImpl.java | 10 +-
.../tier/sockets/OriginalServerConnection.java | 7 +-
.../tier/sockets/ProtobufServerConnection.java | 7 +-
.../cache/tier/sockets/ServerConnection.java | 1 +
.../tier/sockets/ServerConnectionFactory.java | 24 +-
.../internal/cache/wan/AbstractGatewaySender.java | 7 +-
.../geode/statistics/cache/CacheServerStats.kt | 248 ++++----
.../statistics/distributed/DistributionStats.kt | 7 +-
.../distributed/ThrottledMemQueueStatHelper.kt | 52 --
.../apache/geode/statistics/dlock/DLockStats.kt | 651 +++++++++++++++++++++
.../statistics/dlock/DistributedLockStats.java | 270 +++++++++
.../geode/statistics/wan/AsyncEventQueueStats.java | 186 ++++++
.../geode/statistics/wan/GatewayReceiverStats.kt | 131 +++++
.../geode/statistics/wan/GatewaySenderStats.kt | 374 ++++++++++++
39 files changed, 2294 insertions(+), 713 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/Bag.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/Bag.java
index 9ef2999..ab3e098 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/Bag.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/Bag.java
@@ -33,8 +33,8 @@ import org.apache.geode.cache.query.internal.types.CollectionTypeImpl;
import org.apache.geode.cache.query.types.CollectionType;
import org.apache.geode.cache.query.types.ObjectType;
import org.apache.geode.cache.query.types.StructType;
-import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.i18n.LocalizedStrings;
+import org.apache.geode.statistics.cache.CachePerfStats;
// @todo probably should assert element type when elements added
// @todo support generics when no longer support Java 1.4
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/ExecutionContext.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/ExecutionContext.java
index a821fb4..cbc5575 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/ExecutionContext.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/ExecutionContext.java
@@ -35,11 +35,11 @@ import org.apache.geode.cache.query.internal.parse.OQLLexerTokenTypes;
import org.apache.geode.cache.query.internal.types.TypeUtils;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.BucketRegion;
-import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.pdx.internal.PdxString;
+import org.apache.geode.statistics.cache.CachePerfStats;
/**
* This is used to carry the state of a query or index update operation. A state of a query is a set
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/ResultsBag.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/ResultsBag.java
index 2baa5ae..3041fa5 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/ResultsBag.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/ResultsBag.java
@@ -26,7 +26,7 @@ import org.apache.geode.cache.query.internal.ObjectIntHashMap.Entry;
import org.apache.geode.cache.query.types.ObjectType;
import org.apache.geode.internal.DataSerializableFixedID;
import org.apache.geode.internal.Version;
-import org.apache.geode.internal.cache.CachePerfStats;
+import org.apache.geode.statistics.cache.CachePerfStats;
public class ResultsBag extends Bag implements DataSerializableFixedID {
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterDistributionManager.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterDistributionManager.java
index 208788a..4dbc172 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterDistributionManager.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/ClusterDistributionManager.java
@@ -92,13 +92,14 @@ import org.apache.geode.internal.tcp.Connection;
import org.apache.geode.internal.tcp.ConnectionTable;
import org.apache.geode.internal.tcp.ReenteredConnectException;
import org.apache.geode.internal.util.concurrent.StoppableReentrantLock;
+import org.apache.geode.statistics.distributed.DMStats;
+import org.apache.geode.statistics.distributed.DistributionStats;
/**
- * The <code>DistributionManager</code> uses a {@link MembershipManager} to distribute
- * {@link DistributionMessage messages}. It also reports on who is currently in the distributed
- * system and tracks the elder member for the distributed lock service. You may also register a
- * membership listener with the DistributionManager to receive notification of changes in
- * membership.
+ * The <code>DistributionManager</code> uses a {@link MembershipManager} to distribute {@link
+ * DistributionMessage messages}. It also reports on who is currently in the distributed system and
+ * tracks the elder member for the distributed lock service. You may also register a membership
+ * listener with the DistributionManager to receive notification of changes in membership.
*
* <P>
*
@@ -106,7 +107,6 @@ import org.apache.geode.internal.util.concurrent.StoppableReentrantLock;
* <code>DistributionManager</code> and invoke {@link #putOutgoing}.
*
* <P>
- *
* @see DistributionMessage#process
* @see IgnoredByManager
*/
@@ -136,7 +136,6 @@ public class ClusterDistributionManager implements DistributionManager {
private static final int MAX_STOP_ATTEMPTS = 10;
-
private static final boolean SYNC_EVENTS = Boolean.getBoolean("DistributionManager.syncEvents");
/**
@@ -162,11 +161,12 @@ public class ClusterDistributionManager implements DistributionManager {
Math.max(Runtime.getRuntime().availableProcessors() * 4, 16)).intValue();
-
private static final int INCOMING_QUEUE_LIMIT =
Integer.getInteger("DistributionManager.INCOMING_QUEUE_LIMIT", 80000).intValue();
- /** Throttling based on the Queue byte size */
+ /**
+ * Throttling based on the Queue byte size
+ */
private static final double THROTTLE_PERCENT = (double) (Integer
.getInteger("DistributionManager.SERIAL_QUEUE_THROTTLE_PERCENT", 75).intValue()) / 100;
@@ -185,7 +185,9 @@ public class ClusterDistributionManager implements DistributionManager {
Integer.getInteger("DistributionManager.TOTAL_SERIAL_QUEUE_THROTTLE",
(int) (SERIAL_QUEUE_BYTE_LIMIT * THROTTLE_PERCENT)).intValue();
- /** Throttling based on the Queue item size */
+ /**
+ * Throttling based on the Queue item size
+ */
static final int SERIAL_QUEUE_SIZE_LIMIT =
Integer.getInteger("DistributionManager.SERIAL_QUEUE_SIZE_LIMIT", 20000).intValue();
@@ -193,26 +195,34 @@ public class ClusterDistributionManager implements DistributionManager {
Integer.getInteger("DistributionManager.SERIAL_QUEUE_SIZE_THROTTLE",
(int) (SERIAL_QUEUE_SIZE_LIMIT * THROTTLE_PERCENT)).intValue();
- /** Max number of serial Queue executors, in case of multi-serial-queue executor */
+ /**
+ * Max number of serial Queue executors, in case of multi-serial-queue executor
+ */
static final int MAX_SERIAL_QUEUE_THREAD =
Integer.getInteger("DistributionManager.MAX_SERIAL_QUEUE_THREAD", 20).intValue();
-
- /** The DM type for regular distribution managers */
+ /**
+ * The DM type for regular distribution managers
+ */
public static final int NORMAL_DM_TYPE = 10;
- /** The DM type for locator distribution managers */
+ /**
+ * The DM type for locator distribution managers
+ */
public static final int LOCATOR_DM_TYPE = 11;
- /** The DM type for Console (admin-only) distribution managers */
+ /**
+ * The DM type for Console (admin-only) distribution managers
+ */
public static final int ADMIN_ONLY_DM_TYPE = 12;
- /** The DM type for stand-alone members */
+ /**
+ * The DM type for stand-alone members
+ */
public static final int LONER_DM_TYPE = 13;
-
/**
* @see org.apache.geode.distributed.internal.PooledDistributionMessage
*/
@@ -243,7 +253,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Executor for view related messages
- *
* @see org.apache.geode.distributed.internal.membership.gms.messages.ViewAckMessage
*/
public static final int VIEW_EXECUTOR = 79;
@@ -252,8 +261,9 @@ public class ClusterDistributionManager implements DistributionManager {
public static final int REGION_FUNCTION_EXECUTION_EXECUTOR = 80;
-
- /** Is this node running an AdminDistributedSystem? */
+ /**
+ * Is this node running an AdminDistributedSystem?
+ */
private static volatile boolean isDedicatedAdminVM = false;
private static ThreadLocal<Boolean> isStartupThread = new ThreadLocal();
@@ -269,8 +279,6 @@ public class ClusterDistributionManager implements DistributionManager {
}
};
-
-
///////////////////// Instance Fields //////////////////////
/**
@@ -280,7 +288,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Must be read/written while holding {@link #elderMonitor}
- *
* @see #elderChangeWait()
*/
private boolean waitingForElderChange = false;
@@ -295,12 +302,13 @@ public class ClusterDistributionManager implements DistributionManager {
* arbitration.
*
* Must hold {@link #elderMonitor} in order to change this.
- *
* @see #getElderId()
*/
protected volatile InternalDistributedMember elder = null;
- /** The id of this distribution manager */
+ /**
+ * The id of this distribution manager
+ */
protected final InternalDistributedMember localAddress;
/**
@@ -315,37 +323,47 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* The <code>MembershipListener</code>s that are registered on this manager for ALL members.
- *
* @since GemFire 5.7
*/
protected volatile Set<MembershipListener> allMembershipListeners = Collections.emptySet();
/**
* A lock to hold while adding and removing all membership listeners.
- *
* @since GemFire 5.7
*/
private final Object allMembershipListenersLock = new MembershipListenersLock();
- /** A queue of MemberEvent instances */
+ /**
+ * A queue of MemberEvent instances
+ */
private final BlockingQueue<MemberEvent> membershipEventQueue = new LinkedBlockingQueue();
- /** Used to invoke registered membership listeners in the background. */
+ /**
+ * Used to invoke registered membership listeners in the background.
+ */
private Thread memberEventThread;
- /** A brief description of this DistributionManager */
+ /**
+ * A brief description of this DistributionManager
+ */
protected final String description;
- /** Statistics about distribution */
+ /**
+ * Statistics about distribution
+ */
protected DistributionStats stats;
- /** Did an exception occur in one of the DM threads? */
+ /**
+ * Did an exception occur in one of the DM threads?
+ */
private boolean exceptionInThreads;
private volatile boolean shutdownMsgSent = false;
- /** Set to true when this manager is being shutdown */
+ /**
+ * Set to true when this manager is being shutdown
+ */
private volatile boolean closeInProgress = false;
private volatile boolean receivedStartupResponse = false;
@@ -383,11 +401,9 @@ public class ClusterDistributionManager implements DistributionManager {
Collections.emptyMap();
/**
- * The lock held while accessing the field references to the following:<br>
- * 1) {@link #members}<br>
- * 2) {@link #membersAndAdmin}<br>
- * 3) {@link #hostedLocatorsAll}<br>
- * 4) {@link #hostedLocatorsWithSharedConfiguration}<br>
+ * The lock held while accessing the field references to the following:<br> 1) {@link
+ * #members}<br> 2) {@link #membersAndAdmin}<br> 3) {@link #hostedLocatorsAll}<br> 4) {@link
+ * #hostedLocatorsWithSharedConfiguration}<br>
*/
private final Object membersLock = new MembersLock();
@@ -401,10 +417,14 @@ public class ClusterDistributionManager implements DistributionManager {
*/
private volatile Set<InternalDistributedMember> adminConsoles = Collections.emptySet();
- /** The group of distribution manager threads */
+ /**
+ * The group of distribution manager threads
+ */
protected LoggingThreadGroup threadGroup;
- /** Message processing thread pool */
+ /**
+ * Message processing thread pool
+ */
private ThreadPoolExecutor threadPool;
/**
@@ -426,22 +446,24 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Thread used to decouple {@link org.apache.geode.internal.cache.partitioned.PartitionMessage}s
* from {@link org.apache.geode.internal.cache.DistributedCacheOperation}s </b>
- *
* @see #SERIAL_EXECUTOR
*/
private ThreadPoolExecutor partitionedRegionThread;
private ThreadPoolExecutor partitionedRegionPool;
- /** Function Execution executors */
+ /**
+ * Function Execution executors
+ */
private ThreadPoolExecutor functionExecutionThread;
private ThreadPoolExecutor functionExecutionPool;
- /** Message processing executor for serial, ordered, messages. */
+ /**
+ * Message processing executor for serial, ordered, messages.
+ */
private ThreadPoolExecutor serialThread;
/**
* Message processing executor for view messages
- *
* @see org.apache.geode.distributed.internal.membership.gms.messages.ViewAckMessage
*/
private ThreadPoolExecutor viewThread;
@@ -454,12 +476,13 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Thread Monitor mechanism to monitor system threads
- *
* @see org.apache.geode.internal.monitoring.ThreadsMonitoring
*/
private final ThreadsMonitoring threadMonitor;
- /** a map keyed on InternalDistributedMember, to direct channels to other systems */
+ /**
+ * a map keyed on InternalDistributedMember, to direct channels to other systems
+ */
// protected final Map channelMap = CFactory.createCM();
private volatile boolean readyForMessages = false;
@@ -471,7 +494,9 @@ public class ClusterDistributionManager implements DistributionManager {
private volatile boolean readyToSendMsgs = false;
private final Object readyToSendMsgsLock = new Object();
- /** Is this distribution manager closed? */
+ /**
+ * Is this distribution manager closed?
+ */
private volatile boolean closed = false;
/**
@@ -479,7 +504,9 @@ public class ClusterDistributionManager implements DistributionManager {
*/
private InternalDistributedSystem system;
- /** The remote transport configuration for this dm */
+ /**
+ * The remote transport configuration for this dm
+ */
private RemoteTransportConfig transport;
/**
@@ -519,8 +546,6 @@ public class ClusterDistributionManager implements DistributionManager {
*/
private final Object shutdownMutex = new Object();
-
-
////////////////////// Static Methods //////////////////////
/**
@@ -533,7 +558,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Creates a new distribution manager and discovers the other members of the distributed system.
* Note that it does not check to see whether or not this VM already has a distribution manager.
- *
* @param system The distributed system to which this distribution manager will send messages.
*/
static ClusterDistributionManager create(InternalDistributedSystem system) {
@@ -627,7 +651,7 @@ public class ClusterDistributionManager implements DistributionManager {
if (logger.isInfoEnabled()) {
long delta = System.currentTimeMillis() - start;
- Object[] logArgs = new Object[] {distributionManager.getDistributionManagerId(), transport,
+ Object[] logArgs = new Object[]{distributionManager.getDistributionManagerId(), transport,
Integer.valueOf(distributionManager.getOtherDistributionManagerIds().size()),
distributionManager.getOtherDistributionManagerIds(),
(logger.isInfoEnabled(LogMarker.DM_MARKER) ? " (VERBOSE, took " + delta + " ms)" : ""),
@@ -657,12 +681,10 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Creates a new <code>DistributionManager</code> by initializing itself, creating the membership
* manager and executors
- *
* @param transport The configuration for the communications transport
- *
*/
private ClusterDistributionManager(RemoteTransportConfig transport,
- InternalDistributedSystem system) {
+ InternalDistributedSystem system) {
this.dmType = transport.getVmKind();
this.system = system;
@@ -672,13 +694,7 @@ public class ClusterDistributionManager implements DistributionManager {
this.membershipListeners = new ConcurrentHashMap();
this.distributedSystemId = system.getConfig().getDistributedSystemId();
{
- long statId = OSProcess.getId();
- /*
- * deadcoded since we don't know the channel id yet. if (statId == 0 || statId == -1) { statId
- * = getMembershipPort(); }
- */
- this.stats = new DistributionStats(system.getStatisticsFactory(), statId);
- DistributionStats.enableClockStats = system.getConfig().getEnableTimeStatistics();
+ this.stats = new DistributionStats((long) OSProcess.getId());
}
this.exceptionInThreads = false;
@@ -832,7 +848,6 @@ public class ClusterDistributionManager implements DistributionManager {
this.threadPool = pool;
}
-
{
BlockingQueue poolQueue;
if (INCOMING_QUEUE_LIMIT == 0) {
@@ -872,7 +887,6 @@ public class ClusterDistributionManager implements DistributionManager {
this.stats.getHighPriorityPoolHelper(), tf, threadMonitor);
}
-
{
ThreadFactory tf = new ThreadFactory() {
private int next = 0;
@@ -1062,7 +1076,7 @@ public class ClusterDistributionManager implements DistributionManager {
logger.info(LocalizedMessage.create(
LocalizedStrings.DistributionManager_STARTING_DISTRIBUTIONMANAGER_0_1,
- new Object[] {this.localAddress,
+ new Object[]{this.localAddress,
(logger.isInfoEnabled(LogMarker.DM_MARKER) ? sb.toString() : "")}));
this.description = "Distribution manager on " + this.localAddress + " started at "
@@ -1078,11 +1092,10 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Creates a new distribution manager
- *
* @param system The distributed system to which this distribution manager will send messages.
*/
private ClusterDistributionManager(InternalDistributedSystem system,
- RemoteTransportConfig transport) {
+ RemoteTransportConfig transport) {
this(transport, system);
boolean finishedConstructor = false;
@@ -1094,7 +1107,7 @@ public class ClusterDistributionManager implements DistributionManager {
// Allow events to start being processed.
membershipManager.startEventProcessing();
- for (;;) {
+ for (; ; ) {
this.getCancelCriterion().checkCancelInProgress(null);
boolean interrupted = Thread.interrupted();
try {
@@ -1120,7 +1133,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Is this VM dedicated to administration (like a GUI console or a JMX agent)? If so, then it
* creates {@link #ADMIN_ONLY_DM_TYPE} type distribution managers.
- *
* @since GemFire 4.0
*/
public static boolean isDedicatedAdminVM() {
@@ -1167,13 +1179,12 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Returns true if the two members are on the same equivalent host based on overlapping IP
* addresses collected for all NICs during exchange of startup messages.
- *
* @param member1 First member
* @param member2 Second member
*/
@Override
public boolean areOnEquivalentHost(InternalDistributedMember member1,
- InternalDistributedMember member2) {
+ InternalDistributedMember member2) {
Set<InetAddress> equivalents1 = getEquivalents(member1.getInetAddress());
return equivalents1.contains(member2.getInetAddress());
}
@@ -1181,7 +1192,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Set the host equivalencies for a given host. This overrides any previous information in the
* tables.
- *
* @param equivs list of InetAddress's that all point at same host
*/
void setEquivalentHosts(Set<InetAddress> equivs) {
@@ -1196,7 +1206,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Return all of the InetAddress's that are equivalent to the given one (same host)
- *
* @param in host to match up
* @return all the addresses thus equivalent
*/
@@ -1220,7 +1229,7 @@ public class ClusterDistributionManager implements DistributionManager {
}
if (member != getDistributionManagerId()) {
String relationship = areInSameZone(getDistributionManagerId(), member) ? "" : "not ";
- Object[] logArgs = new Object[] {member, relationship};
+ Object[] logArgs = new Object[]{member, relationship};
logger.info(LocalizedMessage.create(
LocalizedStrings.DistributionManager_DISTRIBUTIONMANAGER_MEMBER_0_IS_1_EQUIVALENT,
logArgs));
@@ -1308,8 +1317,9 @@ public class ClusterDistributionManager implements DistributionManager {
*/
private void startThreads() {
this.system.setDM(this); // fix for bug 33362
- if (this.memberEventThread != null)
+ if (this.memberEventThread != null) {
this.memberEventThread.start();
+ }
try {
// And the distinguished guests today are...
@@ -1366,12 +1376,14 @@ public class ClusterDistributionManager implements DistributionManager {
}
private void waitUntilReadyForMessages() {
- if (readyForMessages)
+ if (readyForMessages) {
return;
+ }
synchronized (this) {
- for (;;) {
- if (readyForMessages)
+ for (; ; ) {
+ if (readyForMessages) {
break;
+ }
stopper.checkCancelInProgress(null);
boolean interrupted = Thread.interrupted();
try {
@@ -1400,7 +1412,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Return when DM is ready to send out messages.
- *
* @param msg the messsage that is currently being sent
*/
private void waitUntilReadyToSendMsgs(DistributionMessage msg) {
@@ -1421,9 +1432,10 @@ public class ClusterDistributionManager implements DistributionManager {
}
synchronized (this.readyToSendMsgsLock) {
- for (;;) {
- if (this.readyToSendMsgs)
+ for (; ; ) {
+ if (this.readyToSendMsgs) {
break;
+ }
stopper.checkCancelInProgress(null);
boolean interrupted = Thread.interrupted();
try {
@@ -1518,12 +1530,11 @@ public class ClusterDistributionManager implements DistributionManager {
* Adds the entry in {@link #hostedLocatorsAll} for a member with one or more hosted locators. The
* value is a collection of host[port] strings. If a bind-address was used for a locator then the
* form is bind-addr[port].
- *
* @since GemFire 6.6.3
*/
@Override
public void addHostedLocators(InternalDistributedMember member, Collection<String> locators,
- boolean isSharedConfigurationEnabled) {
+ boolean isSharedConfigurationEnabled) {
synchronized (this.membersLock) {
if (locators == null || locators.isEmpty()) {
throw new IllegalArgumentException("Cannot use empty collection of locators");
@@ -1585,12 +1596,10 @@ public class ClusterDistributionManager implements DistributionManager {
}
-
/**
* Gets the value in {@link #hostedLocatorsAll} for a member with one or more hosted locators. The
* value is a collection of host[port] strings. If a bind-address was used for a locator then the
* form is bind-addr[port].
- *
* @since GemFire 6.6.3
*/
@Override
@@ -1612,7 +1621,6 @@ public class ClusterDistributionManager implements DistributionManager {
* The keyset of the map are the locator vms in this cluster.
*
* the value is a collection of strings in case one vm can have multiple locators ????
- *
* @since GemFire 6.6.3
*/
@Override
@@ -1626,7 +1634,6 @@ public class ClusterDistributionManager implements DistributionManager {
* Returns a copy of the map of all members hosting locators with shared configuration. The key is
* the member, and the value is a collection of host[port] strings. If a bind-address was used for
* a locator then the form is bind-addr[port].
- *
* @since GemFire 8.0
*/
@Override
@@ -1775,7 +1782,7 @@ public class ClusterDistributionManager implements DistributionManager {
: "");
logger.info(LocalizedMessage.create(
LocalizedStrings.DistributionManager_SHUTTING_DOWN_DISTRIBUTIONMANAGER_0_1,
- new Object[] {this.localAddress, exceptionStatus}));
+ new Object[]{this.localAddress, exceptionStatus}));
final long start = System.currentTimeMillis();
try {
@@ -1890,15 +1897,16 @@ public class ClusterDistributionManager implements DistributionManager {
}
Thread th = this.memberEventThread;
- if (th != null)
+ if (th != null) {
th.interrupt();
+ }
}
private void waitForThreadsToStop(long timeInMillis) throws InterruptedException {
long start = System.currentTimeMillis();
long remaining = timeInMillis;
- ExecutorService[] allExecutors = new ExecutorService[] {this.serialThread, this.viewThread,
+ ExecutorService[] allExecutors = new ExecutorService[]{this.serialThread, this.viewThread,
this.functionExecutionThread, this.functionExecutionPool, this.partitionedRegionThread,
this.partitionedRegionPool, this.highPriorityPool, this.waitingPool,
this.prMetaDataCleanupThreadPool, this.threadPool};
@@ -1912,7 +1920,6 @@ public class ClusterDistributionManager implements DistributionManager {
}
}
-
this.serialQueuedExecutorPool.awaitTermination(remaining, TimeUnit.MILLISECONDS);
remaining = timeInMillis - (System.currentTimeMillis() - start);
if (remaining <= 0) {
@@ -1921,7 +1928,7 @@ public class ClusterDistributionManager implements DistributionManager {
Thread th = this.memberEventThread;
if (th != null) {
th.interrupt(); // bug #43452 - this thread sometimes eats interrupts, so we interrupt it
- // again here
+ // again here
th.join(remaining);
}
@@ -1929,12 +1936,12 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Cheap tool to kill a referenced thread
- *
* @param t the thread to kill
*/
private void clobberThread(Thread t) {
- if (t == null)
+ if (t == null) {
return;
+ }
if (t.isAlive()) {
logger.warn(LocalizedMessage
.create(LocalizedStrings.DistributionManager_FORCING_THREAD_STOP_ON__0_, t));
@@ -1964,7 +1971,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Cheap tool to examine an executor to see if it is still working
- *
* @return true if executor is still active
*/
private boolean executorAlive(ThreadPoolExecutor tpe, String name) {
@@ -1986,12 +1992,11 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Wait for the ancillary queues to exit. Kills them if they are still around.
- *
*/
private void forceThreadsToStop() {
long endTime = System.currentTimeMillis() + MAX_STOP_TIME;
String culprits = "";
- for (;;) {
+ for (; ; ) {
boolean stillAlive = false;
culprits = "";
if (executorAlive(this.serialThread, "serial thread")) {
@@ -2027,12 +2032,14 @@ public class ClusterDistributionManager implements DistributionManager {
culprits = culprits + " thread pool;";
}
- if (!stillAlive)
+ if (!stillAlive) {
return;
+ }
long now = System.currentTimeMillis();
- if (now >= endTime)
+ if (now >= endTime) {
break;
+ }
try {
Thread.sleep(STOP_PAUSE_TIME);
@@ -2090,10 +2097,14 @@ public class ClusterDistributionManager implements DistributionManager {
private volatile boolean shutdownInProgress = false;
- /** guard for membershipViewIdAcknowledged */
+ /**
+ * guard for membershipViewIdAcknowledged
+ */
private final Object membershipViewIdGuard = new Object();
- /** the latest view ID that has been processed by all membership listeners */
+ /**
+ * the latest view ID that has been processed by all membership listeners
+ */
private long membershipViewIdAcknowledged;
@Override
@@ -2127,23 +2138,11 @@ public class ClusterDistributionManager implements DistributionManager {
} // try
finally {
// ABSOLUTELY ESSENTIAL that we close the distribution channel!
- try {
- // For safety, but channel close in a finally AFTER this...
- if (this.stats != null) {
- this.stats.close();
- try {
- Thread.sleep(100);
- } catch (InterruptedException ie) {
- // No need to reset interrupt bit, we're really trying to quit...
- }
- }
- } finally {
- if (this.membershipManager != null) {
- logger.info(LocalizedMessage.create(
- LocalizedStrings.DistributionManager_NOW_CLOSING_DISTRIBUTION_FOR__0,
- this.localAddress));
- this.membershipManager.disconnect(beforeJoined);
- }
+ if (this.membershipManager != null) {
+ logger.info(LocalizedMessage.create(
+ LocalizedStrings.DistributionManager_NOW_CLOSING_DISTRIBUTION_FOR__0,
+ this.localAddress));
+ this.membershipManager.disconnect(beforeJoined);
}
}
}
@@ -2248,11 +2247,9 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* This thread processes member events as they occur.
- *
* @see ClusterDistributionManager.MemberCrashedEvent
* @see ClusterDistributionManager.MemberJoinedEvent
* @see ClusterDistributionManager.MemberDepartedEvent
- *
*/
protected class MemberEventInvoker implements Runnable {
@@ -2260,7 +2257,7 @@ public class ClusterDistributionManager implements DistributionManager {
@Override
@SuppressWarnings("synthetic-access")
public void run() {
- for (;;) {
+ for (; ; ) {
SystemFailure.checkFailure();
// bug 41539 - member events need to be delivered during shutdown
// or reply processors may hang waiting for replies from
@@ -2397,8 +2394,8 @@ public class ClusterDistributionManager implements DistributionManager {
@Override
public void retainMembersWithSameOrNewerVersion(Collection<InternalDistributedMember> members,
- Version version) {
- for (Iterator<InternalDistributedMember> it = members.iterator(); it.hasNext();) {
+ Version version) {
+ for (Iterator<InternalDistributedMember> it = members.iterator(); it.hasNext(); ) {
InternalDistributedMember id = it.next();
if (id.getVersionObject().compareTo(version) < 0) {
it.remove();
@@ -2408,8 +2405,8 @@ public class ClusterDistributionManager implements DistributionManager {
@Override
public void removeMembersWithSameOrNewerVersion(Collection<InternalDistributedMember> members,
- Version version) {
- for (Iterator<InternalDistributedMember> it = members.iterator(); it.hasNext();) {
+ Version version) {
+ for (Iterator<InternalDistributedMember> it = members.iterator(); it.hasNext(); ) {
InternalDistributedMember id = it.next();
if (id.getVersionObject().compareTo(version) >= 0) {
it.remove();
@@ -2440,8 +2437,9 @@ public class ClusterDistributionManager implements DistributionManager {
*/
private boolean sendStartupMessage(StartupOperation startupOperation)
throws InterruptedException {
- if (Thread.interrupted())
+ if (Thread.interrupted()) {
throw new InterruptedException();
+ }
this.receivedStartupResponse = false;
boolean ok = false;
@@ -2513,10 +2511,11 @@ public class ClusterDistributionManager implements DistributionManager {
int unresponsiveCount;
synchronized (unfinishedStartupsLock) {
- if (unfinishedStartups == null)
+ if (unfinishedStartups == null) {
unresponsiveCount = 0;
- else
+ } else {
unresponsiveCount = unfinishedStartups.size();
+ }
if (unresponsiveCount != 0) {
if (Boolean.getBoolean("DistributionManager.requireAllStartupResponses")) {
@@ -2527,7 +2526,6 @@ public class ClusterDistributionManager implements DistributionManager {
}
} // synchronized
-
// Bug 35887:
// If there are other members, we must receive at least _one_ response
if (allOthers.size() != 0) { // there exist others
@@ -2537,8 +2535,9 @@ public class ClusterDistributionManager implements DistributionManager {
while (itt.hasNext()) {
Object m = itt.next();
sb.append(m.toString());
- if (itt.hasNext())
+ if (itt.hasNext()) {
sb.append(", ");
+ }
}
if (DEBUG_NO_ACKNOWLEDGEMENTS) {
printStacks(allOthers, false);
@@ -2546,7 +2545,7 @@ public class ClusterDistributionManager implements DistributionManager {
throw new SystemConnectException(
LocalizedStrings.DistributionManager_RECEIVED_NO_CONNECTION_ACKNOWLEDGMENTS_FROM_ANY_OF_THE_0_SENIOR_CACHE_MEMBERS_1
.toLocalizedString(
- new Object[] {Integer.toString(allOthers.size()), sb.toString()}));
+ new Object[]{Integer.toString(allOthers.size()), sb.toString()}));
} // and none responded
} // there exist others
@@ -2554,10 +2553,11 @@ public class ClusterDistributionManager implements DistributionManager {
if (e != null) { // an elder exists
boolean unresponsiveElder;
synchronized (unfinishedStartupsLock) {
- if (unfinishedStartups == null)
+ if (unfinishedStartups == null) {
unresponsiveElder = false;
- else
+ } else {
unresponsiveElder = unfinishedStartups.contains(e);
+ }
}
if (unresponsiveElder) {
logger.warn(LocalizedMessage.create(
@@ -2609,10 +2609,12 @@ public class ClusterDistributionManager implements DistributionManager {
if (logger.isDebugEnabled()) {
logger.debug("removeUnfinishedStartup for {} wtih {}", m, unfinishedStartups);
}
- if (unfinishedStartups == null)
+ if (unfinishedStartups == null) {
return; // not yet done with startup
- if (!unfinishedStartups.remove(m))
+ }
+ if (!unfinishedStartups.remove(m)) {
return;
+ }
StringId msg = null;
if (departed) {
msg =
@@ -2626,14 +2628,13 @@ public class ClusterDistributionManager implements DistributionManager {
if (numLeft != 0) {
logger.info(LocalizedMessage.create(
LocalizedStrings.DistributionManager_STILL_AWAITING_0_RESPONSES_FROM_1,
- new Object[] {Integer.valueOf(numLeft), unfinishedStartups}));
+ new Object[]{Integer.valueOf(numLeft), unfinishedStartups}));
}
} // synchronized
}
/**
* Processes the first startup response.
- *
* @see StartupResponseMessage#process
*/
void processStartupResponse(InternalDistributedMember sender, String theRejectionMessage) {
@@ -2652,7 +2653,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Based on a recent JGroups view, return a member that might be the next elder.
- *
* @return the elder candidate, possibly this VM.
*/
private InternalDistributedMember getElderCandidate() {
@@ -2680,8 +2680,9 @@ public class ClusterDistributionManager implements DistributionManager {
while (it.hasNext()) {
InternalDistributedMember member = it.next();
int managerType = member.getVmKind();
- if (managerType == ADMIN_ONLY_DM_TYPE)
+ if (managerType == ADMIN_ONLY_DM_TYPE) {
continue;
+ }
if (managerType == LOCATOR_DM_TYPE) {
// Fix for #50510 - pre-8.0 members will not let a locator be the elder
@@ -2710,8 +2711,9 @@ public class ClusterDistributionManager implements DistributionManager {
while (it.hasNext()) {
InternalDistributedMember member = it.next();
int managerType = member.getVmKind();
- if (managerType == ADMIN_ONLY_DM_TYPE)
+ if (managerType == ADMIN_ONLY_DM_TYPE) {
continue;
+ }
if (managerType == LOCATOR_DM_TYPE) {
// Fix for #50510 - pre-8.0 members will not let a locator be the elder
@@ -2746,11 +2748,10 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Select a new elder
- *
*/
private void selectElder() {
getSystem().getCancelCriterion().checkCancelInProgress(null); // bug 37884, if DS is
- // disconnecting, throw exception
+ // disconnecting, throw exception
// Once we are the elder, we're stuck until we leave the view.
if (this.localAddress.equals(this.elder)) {
@@ -2849,9 +2850,7 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Makes note of a new distribution manager that has started up in the distributed cache. Invokes
* the appropriately listeners.
- *
* @param theId The id of the distribution manager starting up
- *
*/
private void handleManagerStartup(InternalDistributedMember theId) {
HashMap<InternalDistributedMember, InternalDistributedMember> tmp = null;
@@ -2877,7 +2876,7 @@ public class ClusterDistributionManager implements DistributionManager {
}
logger.info(LocalizedMessage.create(
LocalizedStrings.DistributionManager_ADMITTING_MEMBER_0_NOW_THERE_ARE_1_NONADMIN_MEMBERS,
- new Object[] {theId, Integer.valueOf(tmp.size())}));
+ new Object[]{theId, Integer.valueOf(tmp.size())}));
addMemberEvent(new MemberJoinedEvent(theId));
}
@@ -2894,15 +2893,15 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Makes note of a new console that has started up in the distributed cache.
- *
*/
private void handleConsoleStartup(InternalDistributedMember theId) {
// if we have an all listener then notify it NOW.
HashSet tmp = null;
synchronized (this.membersLock) {
// Note test is under membersLock
- if (membersAndAdmin.contains(theId))
+ if (membersAndAdmin.contains(theId)) {
return; // already accounted for
+ }
// Note we don't modify in place. This allows reader to get snapshots
// without locking.
@@ -2911,7 +2910,7 @@ public class ClusterDistributionManager implements DistributionManager {
this.membersAndAdmin = Collections.unmodifiableSet(tmp);
} // synchronized
- for (Iterator iter = allMembershipListeners.iterator(); iter.hasNext();) {
+ for (Iterator iter = allMembershipListeners.iterator(); iter.hasNext(); ) {
MembershipListener listener = (MembershipListener) iter.next();
listener.memberJoined(this, theId);
}
@@ -2939,14 +2938,12 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Makes note of a console that has shut down.
- *
* @param theId The id of the console shutting down
* @param crashed only true if we detect this id to be gone from a javagroup view
- *
* @see AdminConsoleDisconnectMessage#process
*/
public void handleConsoleShutdown(InternalDistributedMember theId, boolean crashed,
- String reason) {
+ String reason) {
boolean removedConsole = false;
boolean removedMember = false;
synchronized (this.membersLock) {
@@ -2955,9 +2952,10 @@ public class ClusterDistributionManager implements DistributionManager {
// This happens when we have an admin guy colocated with a normal DS.
// In this case we need for the normal DS to shutdown or crash.
if (!this.members.containsKey(theId)) {
- if (logger.isDebugEnabled())
+ if (logger.isDebugEnabled()) {
logger.debug("DistributionManager: removing admin member <{}>; crashed = {}; reason = {}",
theId, crashed, reason);
+ }
Set tmp = new HashSet(this.membersAndAdmin);
if (tmp.remove(theId)) {
// Note we don't modify in place. This allows reader to get snapshots
@@ -2991,7 +2989,7 @@ public class ClusterDistributionManager implements DistributionManager {
}
}
if (removedMember) {
- for (Iterator iter = allMembershipListeners.iterator(); iter.hasNext();) {
+ for (Iterator iter = allMembershipListeners.iterator(); iter.hasNext(); ) {
MembershipListener listener = (MembershipListener) iter.next();
listener.memberDeparted(this, theId, crashed);
}
@@ -3003,7 +3001,7 @@ public class ClusterDistributionManager implements DistributionManager {
} else {
msg = LocalizedStrings.DistributionManager_ADMINISTRATION_MEMBER_AT_0_CLOSED_1;
}
- logger.info(LocalizedMessage.create(msg, new Object[] {theId, reason}));
+ logger.info(LocalizedMessage.create(msg, new Object[]{theId, reason}));
}
redundancyZones.remove(theId);
@@ -3017,7 +3015,7 @@ public class ClusterDistributionManager implements DistributionManager {
@Override
public void handleManagerDeparture(InternalDistributedMember theId, boolean p_crashed,
- String p_reason) {
+ String p_reason) {
boolean crashed = p_crashed;
String reason = p_reason;
@@ -3032,8 +3030,6 @@ public class ClusterDistributionManager implements DistributionManager {
// keep going
}
-
-
int vmType = theId.getVmKind();
if (vmType == ADMIN_ONLY_DM_TYPE) {
removeUnfinishedStartup(theId, true);
@@ -3061,7 +3057,7 @@ public class ClusterDistributionManager implements DistributionManager {
LocalizedStrings.DistributionManager_MEMBER_AT_0_GRACEFULLY_LEFT_THE_DISTRIBUTED_CACHE_1;
addMemberEvent(new MemberDepartedEvent(theId, reason));
}
- logger.info(LocalizedMessage.create(msg, new Object[] {theId, prettifyReason(reason)}));
+ logger.info(LocalizedMessage.create(msg, new Object[]{theId, prettifyReason(reason)}));
// Remove this manager from the serialQueueExecutor.
if (this.serialQueuedExecutorPool != null) {
@@ -3071,7 +3067,7 @@ public class ClusterDistributionManager implements DistributionManager {
}
private void handleManagerSuspect(InternalDistributedMember suspect,
- InternalDistributedMember whoSuspected, String reason) {
+ InternalDistributedMember whoSuspected, String reason) {
if (!isCurrentMember(suspect)) {
return; // fault tolerance
}
@@ -3089,7 +3085,7 @@ public class ClusterDistributionManager implements DistributionManager {
}
private void handleQuorumLost(Set<InternalDistributedMember> failures,
- List<InternalDistributedMember> remaining) {
+ List<InternalDistributedMember> remaining) {
addMemberEvent(new QuorumLostEvent(failures, remaining));
}
@@ -3120,12 +3116,10 @@ public class ClusterDistributionManager implements DistributionManager {
try {
// m.resetTimestamp(); // nanotimers across systems don't match
- long startTime = DistributionStats.getStatTime();
+ long startTime = System.nanoTime();
sendViaMembershipManager(m.getRecipients(), m, this, stats);
this.stats.incSentMessages(1L);
- if (DistributionStats.enableClockStats) {
- stats.incSentMessagesTime(DistributionStats.getStatTime() - startTime);
- }
+ stats.incSentMessagesTime(System.nanoTime() - startTime);
} catch (CancelException e) {
logger.debug("CancelException caught sending shutdown: {}", e.getMessage(), e);
} catch (Exception ex2) {
@@ -3165,40 +3159,32 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Actually does the work of sending a message out over the distribution channel.
- *
* @param message the message to send
* @return list of recipients that did not receive the message because they left the view (null if
- * all received it or it was sent to {@link DistributionMessage#ALL_RECIPIENTS}.
+ * all received it or it was sent to {@link DistributionMessage#ALL_RECIPIENTS}.
* @throws NotSerializableException If <code>message</code> cannot be serialized
*/
Set sendOutgoing(DistributionMessage message) throws NotSerializableException {
- long startTime = DistributionStats.getStatTime();
+ long startTime = System.nanoTime();
Set result = sendViaMembershipManager(message.getRecipients(), message,
ClusterDistributionManager.this, this.stats);
long endTime = 0L;
- if (DistributionStats.enableClockStats) {
- endTime = NanoTimer.getTime();
- }
+ endTime = NanoTimer.getTime();
boolean sentToAll = message.forAll();
if (sentToAll) {
stats.incBroadcastMessages(1L);
- if (DistributionStats.enableClockStats) {
- stats.incBroadcastMessagesTime(endTime - startTime);
- }
+ stats.incBroadcastMessagesTime(endTime - startTime);
}
stats.incSentMessages(1L);
- if (DistributionStats.enableClockStats) {
- stats.incSentMessagesTime(endTime - startTime);
- stats.incDistributeMessageTime(endTime - message.getTimestamp());
- }
+ stats.incSentMessagesTime(endTime - startTime);
+ stats.incDistributeMessageTime(endTime - message.getTimestamp());
return result;
}
-
/**
* @return recipients who did not receive the message
* @throws NotSerializableException If <codE>message</code> cannot be serialized
@@ -3225,13 +3211,15 @@ public class ClusterDistributionManager implements DistributionManager {
logger.fatal(
LocalizedMessage.create(LocalizedStrings.DistributionManager_WHILE_PUSHING_MESSAGE_0_TO_1,
- new Object[] {message, receiver}),
+ new Object[]{message, receiver}),
ex);
- if (message == null || message.forAll())
+ if (message == null || message.forAll()) {
return null;
+ }
result = new HashSet();
- for (int i = 0; i < message.getRecipients().length; i++)
+ for (int i = 0; i < message.getRecipients().length; i++) {
result.add(message.getRecipients()[i]);
+ }
return result;
/*
* if (ex instanceof org.apache.geode.GemFireIpcResourceException) { return; }
@@ -3242,17 +3230,19 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* @return list of recipients who did not receive the message because they left the view (null if
- * all received it or it was sent to {@link DistributionMessage#ALL_RECIPIENTS}).
+ * all received it or it was sent to {@link DistributionMessage#ALL_RECIPIENTS}).
* @throws NotSerializableException If content cannot be serialized
*/
private Set sendViaMembershipManager(InternalDistributedMember[] destinations,
- DistributionMessage content, ClusterDistributionManager dm, DistributionStats stats)
+ DistributionMessage content, ClusterDistributionManager dm,
+ DistributionStats stats)
throws NotSerializableException {
if (membershipManager == null) {
logger.warn(LocalizedMessage.create(
LocalizedStrings.DistributionChannel_ATTEMPTING_A_SEND_TO_A_DISCONNECTED_DISTRIBUTIONMANAGER));
- if (destinations.length == 1 && destinations[0] == DistributionMessage.ALL_RECIPIENTS)
+ if (destinations.length == 1 && destinations[0] == DistributionMessage.ALL_RECIPIENTS) {
return null;
+ }
HashSet result = new HashSet();
for (int i = 0; i < destinations.length; i++) {
result.add(destinations[i]);
@@ -3397,7 +3387,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Waits until elder if newElder or newElder is no longer a member
- *
* @return true if newElder is the elder; false if he is no longer a member or we are the elder.
*/
public boolean waitForElder(final InternalDistributedMember desiredElder) {
@@ -3407,8 +3396,9 @@ public class ClusterDistributionManager implements DistributionManager {
// desiredElder.getVmKind() != DistributionManager.ADMIN_ONLY_DM_TYPE);
synchronized (this.elderMonitor) {
while (true) {
- if (closeInProgress)
+ if (closeInProgress) {
return false;
+ }
InternalDistributedMember currentElder = this.elder;
// Assert.assertTrue(
// currentElder.getVmKind() != DistributionManager.ADMIN_ONLY_DM_TYPE);
@@ -3427,13 +3417,13 @@ public class ClusterDistributionManager implements DistributionManager {
l = new MembershipListener() {
@Override
public void memberJoined(DistributionManager distributionManager,
- InternalDistributedMember theId) {
+ InternalDistributedMember theId) {
// nothing needed
}
@Override
public void memberDeparted(DistributionManager distributionManager,
- InternalDistributedMember theId, boolean crashed) {
+ InternalDistributedMember theId, boolean crashed) {
if (desiredElder.equals(theId)) {
notifyElderChangeWaiters();
}
@@ -3441,21 +3431,25 @@ public class ClusterDistributionManager implements DistributionManager {
@Override
public void memberSuspect(DistributionManager distributionManager,
- InternalDistributedMember id, InternalDistributedMember whoSuspected,
- String reason) {}
+ InternalDistributedMember id,
+ InternalDistributedMember whoSuspected,
+ String reason) {
+ }
- public void viewInstalled(NetView view) {}
+ public void viewInstalled(NetView view) {
+ }
@Override
public void quorumLost(DistributionManager distributionManager,
- Set<InternalDistributedMember> failures,
- List<InternalDistributedMember> remaining) {}
+ Set<InternalDistributedMember> failures,
+ List<InternalDistributedMember> remaining) {
+ }
};
addMembershipListener(l);
}
logger.info(LocalizedMessage.create(
LocalizedStrings.DistributionManager_CHANGING_ELDER_FROM_0_TO_1,
- new Object[] {currentElder, desiredElder}));
+ new Object[]{currentElder, desiredElder}));
elderChangeWait();
} // while true
}
@@ -3570,7 +3564,9 @@ public class ClusterDistributionManager implements DistributionManager {
}
}
- /** returns the serialThread's queue if throttling is being used, null if not */
+ /**
+ * returns the serialThread's queue if throttling is being used, null if not
+ */
public OverflowQueueWithDMStats getSerialQueue(InternalDistributedMember sender) {
if (MULTI_SERIAL_EXECUTORS) {
return this.serialQueuedExecutorPool.getSerialQueue(sender);
@@ -3618,9 +3614,8 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Returns a description of the distribution configuration used for this distribution manager. (in
* ConsoleDistributionManager)
- *
* @return <code>null</code> if no admin {@linkplain #getAgent agent} is associated with this
- * distribution manager
+ * distribution manager
*/
public String getDistributionConfigDescription() {
if (this.agent == null) {
@@ -3638,10 +3633,9 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Returns the health monitor for this distribution manager and owner.
- *
* @param owner the agent that owns the returned monitor
* @return the health monitor created by the owner; <code>null</code> if the owner has now created
- * a monitor.
+ * a monitor.
* @since GemFire 3.5
*/
@Override
@@ -3651,7 +3645,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Returns the health monitor for this distribution manager.
- *
* @param owner the agent that owns the created monitor
* @param cfg the configuration to use when creating the monitor
* @since GemFire 3.5
@@ -3677,7 +3670,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Remove a monitor that was previously created.
- *
* @param owner the agent that owns the monitor to remove
*/
@Override
@@ -3703,14 +3695,16 @@ public class ClusterDistributionManager implements DistributionManager {
return this.adminConsoles;
}
- /** Returns count of members filling the specified role */
+ /**
+ * Returns count of members filling the specified role
+ */
@Override
public int getRoleCount(Role role) {
int count = 0;
Set mbrs = getDistributionManagerIds();
- for (Iterator mbrIter = mbrs.iterator(); mbrIter.hasNext();) {
+ for (Iterator mbrIter = mbrs.iterator(); mbrIter.hasNext(); ) {
Set roles = ((InternalDistributedMember) mbrIter.next()).getRoles();
- for (Iterator rolesIter = roles.iterator(); rolesIter.hasNext();) {
+ for (Iterator rolesIter = roles.iterator(); rolesIter.hasNext(); ) {
Role mbrRole = (Role) rolesIter.next();
if (mbrRole.equals(role)) {
count++;
@@ -3721,13 +3715,15 @@ public class ClusterDistributionManager implements DistributionManager {
return count;
}
- /** Returns true if at least one member is filling the specified role */
+ /**
+ * Returns true if at least one member is filling the specified role
+ */
@Override
public boolean isRolePresent(Role role) {
Set mbrs = getDistributionManagerIds();
- for (Iterator mbrIter = mbrs.iterator(); mbrIter.hasNext();) {
+ for (Iterator mbrIter = mbrs.iterator(); mbrIter.hasNext(); ) {
Set roles = ((InternalDistributedMember) mbrIter.next()).getRoles();
- for (Iterator rolesIter = roles.iterator(); rolesIter.hasNext();) {
+ for (Iterator rolesIter = roles.iterator(); rolesIter.hasNext(); ) {
Role mbrRole = (Role) rolesIter.next();
if (mbrRole.equals(role)) {
return true;
@@ -3737,14 +3733,16 @@ public class ClusterDistributionManager implements DistributionManager {
return false;
}
- /** Returns a set of all roles currently in the distributed system. */
+ /**
+ * Returns a set of all roles currently in the distributed system.
+ */
@Override
public Set getAllRoles() {
Set allRoles = new HashSet();
Set mbrs = getDistributionManagerIds();
- for (Iterator mbrIter = mbrs.iterator(); mbrIter.hasNext();) {
+ for (Iterator mbrIter = mbrs.iterator(); mbrIter.hasNext(); ) {
Set roles = ((InternalDistributedMember) mbrIter.next()).getRoles();
- for (Iterator rolesIter = roles.iterator(); rolesIter.hasNext();) {
+ for (Iterator rolesIter = roles.iterator(); rolesIter.hasNext(); ) {
Role mbrRole = (Role) rolesIter.next();
allRoles.add(mbrRole);
}
@@ -3767,7 +3765,6 @@ public class ClusterDistributionManager implements DistributionManager {
return membershipManager;
}
-
////////////////////// Inner Classes //////////////////////
@@ -3777,13 +3774,19 @@ public class ClusterDistributionManager implements DistributionManager {
* the same thread.
*/
private static class SerialQueuedExecutorPool {
- /** To store the serial threads */
+ /**
+ * To store the serial threads
+ */
ConcurrentMap serialQueuedExecutorMap = new ConcurrentHashMap(MAX_SERIAL_QUEUE_THREAD);
- /** To store the queue associated with thread */
+ /**
+ * To store the queue associated with thread
+ */
Map serialQueuedMap = new HashMap(MAX_SERIAL_QUEUE_THREAD);
- /** Holds mapping between sender to the serial thread-id */
+ /**
+ * Holds mapping between sender to the serial thread-id
+ */
Map senderToSerialQueueIdMap = new HashMap();
/**
@@ -3801,11 +3804,10 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* Constructor.
- *
* @param group thread group to which the threads will belog to.
*/
SerialQueuedExecutorPool(ThreadGroup group, DistributionStats stats,
- boolean throttlingDisabled, ThreadsMonitoring tMonitoring) {
+ boolean throttlingDisabled, ThreadsMonitoring tMonitoring) {
this.threadGroup = group;
this.stats = stats;
this.throttlingDisabled = throttlingDisabled;
@@ -3867,13 +3869,13 @@ public class ClusterDistributionManager implements DistributionManager {
SerialQueuedExecutorWithDMStats executor = getSerialExecutor(sender);
// Get the total serial queue size.
- int totalSerialQueueMemSize = stats.getSerialQueueBytes();
+ int totalSerialQueueMemSize = (int) stats.getSerialQueueBytes();
// for tcp socket reader threads, this code throttles the thread
// to keep the sender-side from overwhelming the receiver.
// UDP readers are throttled in the FC protocol, which queries
// the queue to see if it should throttle
- if (stats.getSerialQueueBytes() > TOTAL_SERIAL_QUEUE_THROTTLE
+ if (totalSerialQueueMemSize > TOTAL_SERIAL_QUEUE_THROTTLE
&& !DistributionMessage.isPreciousThread()) {
do {
boolean interrupted = Thread.interrupted();
@@ -3981,7 +3983,7 @@ public class ClusterDistributionManager implements DistributionManager {
senderToSerialQueueIdMap.remove(member);
// Check if any other members are using the same executor.
- for (Iterator iter = senderToSerialQueueIdMap.values().iterator(); iter.hasNext();) {
+ for (Iterator iter = senderToSerialQueueIdMap.values().iterator(); iter.hasNext(); ) {
Integer value = (Integer) iter.next();
if (value.equals(queueId)) {
isUsed = true;
@@ -3991,11 +3993,12 @@ public class ClusterDistributionManager implements DistributionManager {
// If not used mark this as unused.
if (!isUsed) {
- if (logger.isInfoEnabled(LogMarker.DM_MARKER))
+ if (logger.isInfoEnabled(LogMarker.DM_MARKER)) {
logger.info(LogMarker.DM_MARKER,
LocalizedMessage.create(
LocalizedStrings.DistributionManager_MARKING_THE_SERIALQUEUEDEXECUTOR_WITH_ID__0__USED_BY_THE_MEMBER__1__TO_BE_UNUSED,
- new Object[] {queueId, member}));
+ new Object[]{queueId, member}));
+ }
threadMarkedForUse.add(queueId);
}
@@ -4006,7 +4009,7 @@ public class ClusterDistributionManager implements DistributionManager {
long timeNanos = unit.toNanos(time);
long remainingNanos = timeNanos;
long start = System.nanoTime();
- for (Iterator iter = serialQueuedExecutorMap.values().iterator(); iter.hasNext();) {
+ for (Iterator iter = serialQueuedExecutorMap.values().iterator(); iter.hasNext(); ) {
ExecutorService executor = (ExecutorService) iter.next();
executor.awaitTermination(remainingNanos, TimeUnit.NANOSECONDS);
remainingNanos = timeNanos = (System.nanoTime() - start);
@@ -4017,7 +4020,7 @@ public class ClusterDistributionManager implements DistributionManager {
}
private void shutdown() {
- for (Iterator iter = serialQueuedExecutorMap.values().iterator(); iter.hasNext();) {
+ for (Iterator iter = serialQueuedExecutorMap.values().iterator(); iter.hasNext(); ) {
ExecutorService executor = (ExecutorService) iter.next();
executor.shutdown();
}
@@ -4039,12 +4042,12 @@ public class ClusterDistributionManager implements DistributionManager {
* class so that it shows up nicely in stack traces.
*/
private static class MembershipListenersLock {
- protected MembershipListenersLock() {}
+ protected MembershipListenersLock() {
+ }
}
/**
* This is the listener implementation for responding from events from the Membership Manager.
- *
*/
private class DMListener implements DistributedMembershipListener {
ClusterDistributionManager dm;
@@ -4092,7 +4095,7 @@ public class ClusterDistributionManager implements DistributionManager {
message.setCrashed(crashed);
message.setAlertListenerExpected(true);
message.setIgnoreAlertListenerRemovalFailure(true); // we don't know if it was a listener so
- // don't issue a warning
+ // don't issue a warning
message.setRecipient(localAddress);
message.setReason(reason); // added for #37950
handleIncomingDMsg(message);
@@ -4102,7 +4105,7 @@ public class ClusterDistributionManager implements DistributionManager {
@Override
public void memberSuspect(InternalDistributedMember suspect,
- InternalDistributedMember whoSuspected, String reason) {
+ InternalDistributedMember whoSuspected, String reason) {
dm.handleManagerSuspect(suspect, whoSuspected, reason);
}
@@ -4114,7 +4117,7 @@ public class ClusterDistributionManager implements DistributionManager {
@Override
public void quorumLost(Set<InternalDistributedMember> failures,
- List<InternalDistributedMember> remaining) {
+ List<InternalDistributedMember> remaining) {
dm.handleQuorumLost(failures, remaining);
}
@@ -4153,10 +4156,10 @@ public class ClusterDistributionManager implements DistributionManager {
}
protected abstract void handleEvent(ClusterDistributionManager manager,
- MembershipListener listener);
+ MembershipListener listener);
private void handleEvent(ClusterDistributionManager manager,
- Set<MembershipListener> membershipListeners) {
+ Set<MembershipListener> membershipListeners) {
for (MembershipListener listener : membershipListeners) {
try {
handleEvent(manager, listener);
@@ -4192,8 +4195,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* This is an event reflecting that a InternalDistributedMember has joined the system.
- *
- *
*/
private static class MemberJoinedEvent extends MemberEvent {
MemberJoinedEvent(InternalDistributedMember id) {
@@ -4213,7 +4214,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* This is an event reflecting that a InternalDistributedMember has left the system.
- *
*/
private static class MemberDepartedEvent extends MemberEvent {
String reason;
@@ -4237,8 +4237,6 @@ public class ClusterDistributionManager implements DistributionManager {
/**
* This is an event reflecting that a InternalDistributedMember has left the system in an
* unexpected way.
- *
- *
*/
private static class MemberCrashedEvent extends MemberEvent {
String reason;
@@ -4268,7 +4266,7 @@ public class ClusterDistributionManager implements DistributionManager {
String reason;
MemberSuspectEvent(InternalDistributedMember suspect, InternalDistributedMember whoSuspected,
- String reason) {
+ String reason) {
super(suspect);
this.whoSuspected = whoSuspected;
this.reason = reason;
@@ -4326,7 +4324,7 @@ public class ClusterDistributionManager implements DistributionManager {
List<InternalDistributedMember> remaining;
QuorumLostEvent(Set<InternalDistributedMember> failures,
- List<InternalDistributedMember> remaining) {
+ List<InternalDistributedMember> remaining) {
super(null);
this.failures = failures;
this.remaining = remaining;
@@ -4400,7 +4398,7 @@ public class ClusterDistributionManager implements DistributionManager {
} else {
buddyMembers.add(targetMember);
Set targetAddrs = getEquivalents(targetMember.getInetAddress());
- for (Iterator i = getDistributionManagerIds().iterator(); i.hasNext();) {
+ for (Iterator i = getDistributionManagerIds().iterator(); i.hasNext(); ) {
InternalDistributedMember o = (InternalDistributedMember) i.next();
if (SetUtils.intersectsWith(targetAddrs, getEquivalents(o.getInetAddress()))) {
buddyMembers.add(o);
@@ -4412,7 +4410,7 @@ public class ClusterDistributionManager implements DistributionManager {
@Override
public boolean areInSameZone(InternalDistributedMember member1,
- InternalDistributedMember member2) {
+ InternalDistributedMember member2) {
if (!redundancyZones.isEmpty()) {
String zone1 = redundancyZones.get(member1);
@@ -4460,7 +4458,7 @@ public class ClusterDistributionManager implements DistributionManager {
requiresMessage.addAll(ids);
ids.remove(localAddress);
} else {
- for (Iterator it = ids.iterator(); it.hasNext();) {
+ for (Iterator it = ids.iterator(); it.hasNext(); ) {
InternalDistributedMember mbr = (InternalDistributedMember) it.next();
if (mbr.getProcessId() > 0
&& mbr.getInetAddress().equals(this.localAddress.getInetAddress())) {
@@ -4513,7 +4511,9 @@ public class ClusterDistributionManager implements DistributionManager {
}
}
- /** test method to get the member IDs of all locators in the distributed system */
+ /**
+ * test method to get the member IDs of all locators in the distributed system
+ */
public Set<InternalDistributedMember> getLocatorDistributionManagerIds() {
// access to members synchronized under membersLock in order to
// ensure serialization
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/locks/DLockService.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/locks/DLockService.java
index cf15832..fda73cd 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/locks/DLockService.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/locks/DLockService.java
@@ -36,7 +36,6 @@ import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireException;
-import org.apache.geode.statistics.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedSystem;
@@ -62,10 +61,11 @@ import org.apache.geode.internal.logging.log4j.LocalizedMessage;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.util.StopWatch;
import org.apache.geode.internal.util.concurrent.FutureResult;
+import org.apache.geode.statistics.dlock.DLockStats;
+import org.apache.geode.statistics.dlock.DistributedLockStats;
/**
* Implements the distributed locking service with distributed lock grantors.
- *
*/
public class DLockService extends DistributedLockService {
@@ -86,10 +86,14 @@ public class DLockService extends DistributedLockService {
public static final int INVALID_LEASE_ID = -1;
- /** Unique name for this instance of the named locking service */
+ /**
+ * Unique name for this instance of the named locking service
+ */
protected final String serviceName;
- /** DistributionManager for this member */
+ /**
+ * DistributionManager for this member
+ */
private final DistributionManager dm;
/**
@@ -97,7 +101,9 @@ public class DLockService extends DistributedLockService {
*/
protected final InternalDistributedSystem ds;
- /** Known lock tokens for this service. Key:Object(name), Value:DLockToken */
+ /**
+ * Known lock tokens for this service. Key:Object(name), Value:DLockToken
+ */
private final Map<Object, DLockToken> tokens = new HashMap<Object, DLockToken>();
/**
@@ -112,7 +118,9 @@ public class DLockService extends DistributedLockService {
*/
private final boolean isDistributed;
- /** Optional handler for departure of lease holders; used by grantor */
+ /**
+ * Optional handler for departure of lease holders; used by grantor
+ */
private DLockLessorDepartureHandler lessorDepartureHandler;
/**
@@ -120,7 +128,9 @@ public class DLockService extends DistributedLockService {
*/
private DLockRecoverGrantorProcessor.MessageProcessor recoverGrantorProcessor;
- /** Thread-safe reference to DistributedLockStats */
+ /**
+ * Thread-safe reference to DistributedLockStats
+ */
private final DistributedLockStats dlockStats;
/**
@@ -130,7 +140,9 @@ public class DLockService extends DistributedLockService {
*/
private final Object lockGrantorIdLock = new Object();
- /** Identifies the current grantor for this lock service. */
+ /**
+ * Identifies the current grantor for this lock service.
+ */
private LockGrantorId lockGrantorId;
/**
@@ -146,19 +158,27 @@ public class DLockService extends DistributedLockService {
*/
private int activeLocks = 0;
- /** True if this service should be destroyed in system DisconnectListener */
+ /**
+ * True if this service should be destroyed in system DisconnectListener
+ */
private final boolean destroyOnDisconnect;
- /** True if this service should automatically freeResources */
+ /**
+ * True if this service should automatically freeResources
+ */
private final boolean automateFreeResources;
- /** Identifies the thread that is destroying this lock service. */
+ /**
+ * Identifies the thread that is destroying this lock service.
+ */
private final ThreadLocal<Boolean> destroyingThread = new ThreadLocal<Boolean>();
/// ** Held during destory and creation of this lock service. */
// private final Object serviceLock = new Object();
- /** Protects access to {@link #destroyed} and {@link #activeLocks}. */
+ /**
+ * Protects access to {@link #destroyed} and {@link #activeLocks}.
+ */
private final Object destroyLock = new Object();
/**
@@ -197,7 +217,6 @@ public class DLockService extends DistributedLockService {
/**
* Create a new LockServiceDestroyedException for this lock service.
- *
* @param message the detail message that explains the exception
* @return new LockServiceDestroyedException
*/
@@ -207,7 +226,6 @@ public class DLockService extends DistributedLockService {
/**
* Returns the string message to use in a LockServiceDestroyedException for this lock service.
- *
* @return the detail message that explains LockServiceDestroyedException
*/
protected String generateLockServiceDestroyedMessage() {
@@ -217,7 +235,6 @@ public class DLockService extends DistributedLockService {
/**
* Returns true if {@link #lockGrantorId} is the same as the specified LockGrantorId. Caller must
* synchronize on {@link #lockGrantorIdLock}.
- *
* @param someLockGrantorId the LockGrantorId to check
*/
private boolean checkLockGrantorId(LockGrantorId someLockGrantorId) {
@@ -231,7 +248,6 @@ public class DLockService extends DistributedLockService {
/**
* Returns true if lockGrantorId is the same as the specified LockGrantorId. Caller must
* synchronize on lockGrantorIdLock.
- *
* @param someLockGrantorId the LockGrantorId to check
*/
public boolean isLockGrantorId(LockGrantorId someLockGrantorId) {
@@ -374,14 +390,13 @@ public class DLockService extends DistributedLockService {
* Creates a local {@link DLockGrantor}.
*
* if (!createLocalGrantor(xxx)) { theLockGrantorId = this.lockGrantorId; }
- *
* @param elder the elder that told us to be the grantor
* @param needsRecovery true if recovery is required
* @param myLockGrantorId lockGrantorId to use
* @return true if successfully created local grantor; false if aborted
*/
private boolean createLocalGrantor(InternalDistributedMember elder, boolean needsRecovery,
- LockGrantorId myLockGrantorId) {
+ LockGrantorId myLockGrantorId) {
DLockGrantor myGrantor =
DLockGrantor.createGrantor(this, myLockGrantorId.getLockGrantorVersion());
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
@@ -391,7 +406,7 @@ public class DLockService extends DistributedLockService {
}
private boolean makeLocalGrantor(InternalDistributedMember elder, boolean needsRecovery,
- LockGrantorId myLockGrantorId, DLockGrantor myGrantor) {
+ LockGrantorId myLockGrantorId, DLockGrantor myGrantor) {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
boolean success = false;
try {
@@ -442,11 +457,12 @@ public class DLockService extends DistributedLockService {
// do NOT sync while doing recovery (because it waits for replies)
if (needsRecovery) {
boolean recovered =
- DLockRecoverGrantorProcessor.recoverLockGrantor(this.dm.getDistributionManagerIds(), // include
- // this
- // vm
- this, // this lock service
- myGrantor, this.dm, elder); // the elder that told us to be the grantor
+ DLockRecoverGrantorProcessor
+ .recoverLockGrantor(this.dm.getDistributionManagerIds(), // include
+ // this
+ // vm
+ this, // this lock service
+ myGrantor, this.dm, elder); // the elder that told us to be the grantor
if (!recovered) {
checkDestroyed();
return false; // exit
@@ -536,7 +552,6 @@ public class DLockService extends DistributedLockService {
/**
* Set {@link #lockGrantorId} to the given new value if the current value is null or is an older
* grantor version. Caller must hold {@link #lockGrantorIdLock}.
- *
* @param newLockGrantorId the new value for lockGrantorId
*/
private boolean setLockGrantorId(LockGrantorId newLockGrantorId) {
@@ -566,7 +581,6 @@ public class DLockService extends DistributedLockService {
* Set {@link #lockGrantorId} to the <code>localLockGrantorId</code> if current value is null or
* is an older grantor version. This also atomically sets {@link #grantor} to ensure that the two
* fields are kept in sync. Caller must hold {@link #lockGrantorIdLock}.
- *
* @param localLockGrantorId the new value for lockGrantorId
* @param localGrantor the new local intance of DLockGrantor
*/
@@ -584,7 +598,6 @@ public class DLockService extends DistributedLockService {
/**
* Deposes {@link #lockGrantorId} if <code>newLockGrantorId</code> is newer.
- *
* @param newLockGrantorId the new lock grantor
*/
void deposeOlderLockGrantorId(LockGrantorId newLockGrantorId) {
@@ -612,7 +625,6 @@ public class DLockService extends DistributedLockService {
/**
* Sets {@link #lockGrantorId} to null if the current value equals the expected old value. Caller
* must hold {@link #lockGrantorIdLock}.
- *
* @param oldLockGrantorId the expected old value
* @return true if lockGrantorId was set to null
*/
@@ -654,7 +666,6 @@ public class DLockService extends DistributedLockService {
/**
* Returns true if the grantor version of <code>dlockGrantor</code> equals the
* <code>grantorVersion</code>.
- *
* @param dlockGrantor the grantor instance to compare to grantorVersion
* @param grantorVersion the grantor version number
* @return true if dlockGrantor is the same grantor version
@@ -668,7 +679,6 @@ public class DLockService extends DistributedLockService {
/**
* Returns true if <code>someLockGrantor</code> equals the current {@link #lockGrantorId}.
- *
* @return true if someLockGrantor equals the current lockGrantorId
*/
private boolean equalsLockGrantorId(LockGrantorId someLockGrantor) {
@@ -712,8 +722,8 @@ public class DLockService extends DistributedLockService {
}
/**
- * Increments {@link #activeLocks} while synchronized on {@link #destroyLock} after calling
- * {@link #checkDestroyed()}.
+ * Increments {@link #activeLocks} while synchronized on {@link #destroyLock} after calling {@link
+ * #checkDestroyed()}.
*/
private void incActiveLocks() {
synchronized (this.destroyLock) {
@@ -733,14 +743,13 @@ public class DLockService extends DistributedLockService {
/**
* Returns lockGrantorId when lockGrantorFutureResultRef has been set by another thread.
- *
* @param lockGrantorFutureResultRef FutureResult to wait for
* @param timeToWait how many ms to wait, 0 = forever
* @param timeUnit the unit of measure for timeToWait
* @return the LockGrantorId or null if FutureResult was cancelled
*/
private LockGrantorId waitForLockGrantorFutureResult(FutureResult lockGrantorFutureResultRef,
- long timeToWait, final TimeUnit timeUnit) {
+ long timeToWait, final TimeUnit timeUnit) {
LockGrantorId lockGrantorIdRef = null;
while (lockGrantorIdRef == null) {
boolean interrupted = Thread.interrupted();
@@ -776,12 +785,11 @@ public class DLockService extends DistributedLockService {
/**
* nulls out grantor to force call to elder
- *
* @param timeToWait how long to wait for a new grantor. -1 don't wait, 0 no time limit
* @param timeUnit the unit of measure of timeToWait
*/
private void notLockGrantorId(LockGrantorId notLockGrantorId, long timeToWait,
- final TimeUnit timeUnit) {
+ final TimeUnit timeUnit) {
if (notLockGrantorId.isLocal(getSerialNumber())) {
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE,
@@ -863,7 +871,8 @@ public class DLockService extends DistributedLockService {
}
/**
- * All calls to GrantorRequestProcessor.clearGrantor must come through this synchronization point.
+ * All calls to GrantorRequestProcessor.clearGrantor must come through this synchronization
+ * point.
* <p>
* This fixes a deadlock between this.becomeGrantorMonitor and DistributionManager.elderLock
* <p>
@@ -1035,7 +1044,7 @@ public class DLockService extends DistributedLockService {
lockGrantorFutureResultRef.cancel(true); // interrupt waiting threads
} else {
this.dm.getCancelCriterion().checkCancelInProgress(null); // don't succeed if shutting
- // down
+ // down
// succeeded so set lockGrantorFutureResult
lockGrantorFutureResultRef.set(myLockGrantorId);
}
@@ -1094,7 +1103,6 @@ public class DLockService extends DistributedLockService {
/**
* Attempt to destroy and remove lock token. Synchronizes on tokens map and the lock token.
- *
* @param name the name of the lock token
* @return true if token has been destroyed and removed
*/
@@ -1139,7 +1147,7 @@ public class DLockService extends DistributedLockService {
return;
}
Set unusedTokens = Collections.EMPTY_SET;
- for (Iterator iter = this.tokens.values().iterator(); iter.hasNext();) {
+ for (Iterator iter = this.tokens.values().iterator(); iter.hasNext(); ) {
DLockToken token = (DLockToken) iter.next();
synchronized (token) {
if (!token.isBeingUsed()) {
@@ -1155,7 +1163,7 @@ public class DLockService extends DistributedLockService {
}
}
}
- for (Iterator iter = unusedTokens.iterator(); iter.hasNext();) {
+ for (Iterator iter = unusedTokens.iterator(); iter.hasNext(); ) {
DLockToken token = (DLockToken) iter.next();
synchronized (token) {
int tokensSizeBefore = this.tokens.size();
@@ -1178,7 +1186,7 @@ public class DLockService extends DistributedLockService {
private void removeAllTokens() {
synchronized (this.tokens) {
Assert.assertTrue(this.destroyed);
- for (Iterator iter = this.tokens.values().iterator(); iter.hasNext();) {
+ for (Iterator iter = this.tokens.values().iterator(); iter.hasNext(); ) {
DLockToken token = (DLockToken) iter.next();
synchronized (token) {
token.destroy();
@@ -1194,8 +1202,9 @@ public class DLockService extends DistributedLockService {
checkDestroyed();
synchronized (this.tokens) {
DLockToken token = basicGetToken(name);
- if (token == null)
+ if (token == null) {
return false;
+ }
synchronized (token) {
token.checkForExpiration();
return token.isLeaseHeldByCurrentThread();
@@ -1207,8 +1216,9 @@ public class DLockService extends DistributedLockService {
checkDestroyed();
synchronized (this.tokens) {
DLockToken token = basicGetToken(name);
- if (token == null)
+ if (token == null) {
return false;
+ }
synchronized (token) {
token.checkForExpiration();
if (token.getLesseeThread() == null) {
@@ -1236,13 +1246,13 @@ public class DLockService extends DistributedLockService {
}
public boolean lock(Object name, long waitTimeMillis, long leaseTimeMillis, boolean tryLock,
- boolean disallowReentrant) {
+ boolean disallowReentrant) {
return lock(name, waitTimeMillis, leaseTimeMillis, tryLock, false, false);
}
public boolean lock(Object name, long waitTimeMillis, long leaseTimeMillis, boolean tryLock,
- boolean disallowReentrant, boolean diableAlerts) {
+ boolean disallowReentrant, boolean diableAlerts) {
checkDestroyed();
try {
boolean interruptible = false;
@@ -1267,11 +1277,14 @@ public class DLockService extends DistributedLockService {
return lockInterruptibly(name, waitTimeMillis, leaseTimeMillis, tryLock, interruptible, false);
}
- /** Causes the current thread to sleep for millis and may or may not be interruptible */
+ /**
+ * Causes the current thread to sleep for millis and may or may not be interruptible
+ */
private void sleep(long millis, boolean interruptible) throws InterruptedException {
if (interruptible) {
- if (Thread.interrupted())
+ if (Thread.interrupted()) {
throw new InterruptedException();
+ }
Thread.sleep(millis);
return;
} else {
@@ -1279,7 +1292,9 @@ public class DLockService extends DistributedLockService {
}
}
- /** Causes the current thread to sleep for millis uninterruptibly */
+ /**
+ * Causes the current thread to sleep for millis uninterruptibly
+ */
private void sleep(long millis) {
// Non-interruptible case
StopWatch timer = new StopWatch(true);
@@ -1303,34 +1318,40 @@ public class DLockService extends DistributedLockService {
}
protected DLockRequestProcessor createRequestProcessor(LockGrantorId grantorId, Object name,
- int threadId, long startTime, long requestLeaseTime, long requestWaitTime, boolean reentrant,
- boolean tryLock) {
+ int threadId, long startTime,
+ long requestLeaseTime,
+ long requestWaitTime, boolean reentrant,
+ boolean tryLock) {
return createRequestProcessor(grantorId, name, threadId, startTime, requestLeaseTime,
requestWaitTime, reentrant, tryLock, false);
}
protected DLockRequestProcessor createRequestProcessor(LockGrantorId grantorId, Object name,
- int threadId, long startTime, long requestLeaseTime, long requestWaitTime, boolean reentrant,
- boolean tryLock, boolean disableAlerts) {
+ int threadId, long startTime,
+ long requestLeaseTime,
+ long requestWaitTime, boolean reentrant,
+ boolean tryLock, boolean disableAlerts) {
return new DLockRequestProcessor(grantorId, this, name, threadId, startTime, requestLeaseTime,
requestWaitTime, reentrant, tryLock, disableAlerts, this.dm);
}
protected boolean callReleaseProcessor(InternalDistributedMember grantor, Object name,
- boolean lockBatch, int lockId) {
+ boolean lockBatch, int lockId) {
return DLockService.callReleaseProcessor(this.dm, this.serviceName, grantor, name, lockBatch,
lockId);
}
protected static boolean callReleaseProcessor(DistributionManager dm, String serviceName,
- InternalDistributedMember grantor, Object name, boolean lockBatch, int lockId) {
+ InternalDistributedMember grantor, Object name,
+ boolean lockBatch, int lockId) {
DLockReleaseProcessor processor = new DLockReleaseProcessor(dm, grantor, serviceName, name);
return processor.release(grantor, serviceName, lockBatch, lockId);
}
public boolean lockInterruptibly(final Object name, final long waitTimeMillis,
- final long leaseTimeMillis, final boolean tryLock, final boolean interruptible,
- final boolean disallowReentrant) throws InterruptedException {
+ final long leaseTimeMillis, final boolean tryLock,
+ final boolean interruptible,
+ final boolean disallowReentrant) throws InterruptedException {
return lockInterruptibly(name, waitTimeMillis, leaseTimeMillis, tryLock, interruptible,
disallowReentrant, false);
}
@@ -1338,36 +1359,29 @@ public class DLockService extends DistributedLockService {
/**
* @param name the name of the lock to acquire in this service. This object must conform to the
- * general contract of <code>equals(Object)</code> and <code>hashCode()</code> as described
- * in {@link java.lang.Object#hashCode()}.
- *
+ * general contract of <code>equals(Object)</code> and <code>hashCode()</code> as described in
+ * {@link java.lang.Object#hashCode()}.
* @param waitTimeMillis the number of milliseconds to try to acquire the lock before giving up
- * and returning false. A value of -1 causes this method to block until the lock is
- * acquired.
- *
+ * and returning false. A value of -1 causes this method to block until the lock is acquired.
* @param leaseTimeMillis the number of milliseconds to hold the lock after granting it, before
- * automatically releasing it if it hasn't already been released by invoking
- * {@link #unlock(Object)}. If <code>leaseTimeMillis</code> is -1, hold the lock until
- * explicitly unlocked.
- *
+ * automatically releasing it if it hasn't already been released by invoking {@link
+ * #unlock(Object)}. If <code>leaseTimeMillis</code> is -1, hold the lock until explicitly
+ * unlocked.
* @param tryLock true if the lock should be acquired or fail if currently held. waitTimeMillis
- * will be ignored if the lock is currently held by another client.
- *
+ * will be ignored if the lock is currently held by another client.
* @param interruptible true if this lock request is interruptible
- *
* @param disableAlerts true to disable logging alerts if the dlock is taking a long time to
- * acquired.
- *
+ * acquired.
* @return true if the lock was acquired, false if the timeout <code>waitTimeMillis</code> passed
- * without acquiring the lock.
- *
+ * without acquiring the lock.
* @throws InterruptedException if the thread is interrupted before or during this method.
- *
* @throws UnsupportedOperationException if attempt to lock batch involves non-tryLocks
*/
public boolean lockInterruptibly(final Object name, final long waitTimeMillis,
- final long leaseTimeMillis, final boolean tryLock, final boolean interruptible,
- final boolean disallowReentrant, final boolean disableAlerts) throws InterruptedException {
+ final long leaseTimeMillis, final boolean tryLock,
+ final boolean interruptible,
+ final boolean disallowReentrant, final boolean disableAlerts)
+ throws InterruptedException {
checkDestroyed();
boolean interrupted = Thread.interrupted();
@@ -1383,15 +1397,18 @@ public class DLockService extends DistributedLockService {
long requestLeaseTime = leaseTimeMillis;
// -1 means "lease forever". Long.MAX_VALUE is pretty close.
- if (requestLeaseTime == -1)
+ if (requestLeaseTime == -1) {
requestLeaseTime = Long.MAX_VALUE;
+ }
// -1 means "wait forever". Long.MAX_VALUE is pretty close.
- if (requestWaitTime == -1)
+ if (requestWaitTime == -1) {
requestWaitTime = Long.MAX_VALUE;
+ }
long waitLimit = startTime + requestWaitTime;
- if (waitLimit < 0)
+ if (waitLimit < 0) {
waitLimit = Long.MAX_VALUE;
+ }
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - entering lock()", this, name);
@@ -1440,7 +1457,7 @@ public class DLockService extends DistributedLockService {
if (reentrant && disallowReentrant) {
throw new IllegalStateException(
LocalizedStrings.DLockService_0_ATTEMPTED_TO_REENTER_NONREENTRANT_LOCK_1
- .toLocalizedString(new Object[] {Thread.currentThread(), token}));
+ .toLocalizedString(new Object[]{Thread.currentThread(), token}));
}
recursionBefore = token.getRecursion();
lockId = token.getLeaseId(); // keep lockId
@@ -1479,7 +1496,7 @@ public class DLockService extends DistributedLockService {
}
gotLock = processor.requestLock(interruptible, lockId); // can throw
- // InterruptedException
+ // InterruptedException
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
logger.trace(LogMarker.DLS_VERBOSE, "Grantor {} replied {}", theLockGrantorId,
@@ -1545,7 +1562,7 @@ public class DLockService extends DistributedLockService {
}
} // while (keepTrying)
- // try-block for end stats, token cleanup, and interrupt check
+ // try-block for end stats, token cleanup, and interrupt check
} finally {
getStats().endLockWait(statStart, gotLock);
@@ -1586,8 +1603,11 @@ public class DLockService extends DistributedLockService {
}
private boolean grantLocalDLockAfterObtainingRemoteLock(Object name, DLockToken token,
- int threadId, long leaseExpireTime, int lockId, LockGrantorId theLockGrantorId,
- DLockRequestProcessor processor, int recursion) {
+ int threadId, long leaseExpireTime,
+ int lockId,
+ LockGrantorId theLockGrantorId,
+ DLockRequestProcessor processor,
+ int recursion) {
boolean needToReleaseOrphanedGrant = false;
Assert.assertHoldsLock(this.destroyLock, false);
@@ -1644,11 +1664,9 @@ public class DLockService extends DistributedLockService {
* Suspends granting of locks for this instance of DLockService. If distribute is true, sends
* suspendLocking to all other members that have created this service. Blocks until all
* outstanding locks have been released (excluding those held by the initial calling thread).
- *
* @param waitTimeMillis -1 means "wait forever", >=0 = milliseconds to wait
- *
* @return true if locking is suspended and all locks have been released. Otherwise, resumeLocking
- * is invoked and false is returned.
+ * is invoked and false is returned.
*/
@Override
public boolean suspendLocking(final long waitTimeMillis) {
@@ -1672,8 +1690,9 @@ public class DLockService extends DistributedLockService {
} while (requestWaitTime != 0);
} finally {
- if (interrupted)
+ if (interrupted) {
Thread.currentThread().interrupt();
+ }
}
return false;
@@ -1721,8 +1740,9 @@ public class DLockService extends DistributedLockService {
long startTime = System.currentTimeMillis();
long waitLimit = startTime + waitTimeMillis;
- if (waitLimit < 0)
+ if (waitLimit < 0) {
waitLimit = Long.MAX_VALUE;
+ }
while (!gotToken && keepTrying) {
gotToken =
@@ -1772,7 +1792,7 @@ public class DLockService extends DistributedLockService {
}
throw new LockNotHeldException(
LocalizedStrings.DLockService_ATTEMPTING_TO_UNLOCK_0_1_BUT_THIS_THREAD_DOESNT_OWN_THE_LOCK
- .toLocalizedString(new Object[] {this, name}));
+ .toLocalizedString(new Object[]{this, name}));
}
synchronized (token) {
@@ -1786,7 +1806,7 @@ public class DLockService extends DistributedLockService {
}
throw new LockNotHeldException(
LocalizedStrings.DLockService_ATTEMPTING_TO_UNLOCK_0_1_BUT_THIS_THREAD_DOESNT_OWN_THE_LOCK_2
- .toLocalizedString(new Object[] {this, name, token}));
+ .toLocalizedString(new Object[]{this, name, token}));
}
// if recursion > 0 then token will still be locked after calling release
hadRecursion = token.getRecursion() > 0;
@@ -1854,7 +1874,6 @@ public class DLockService extends DistributedLockService {
/**
* Query the grantor for current leasing information of a lock. Returns the current lease info.
- *
* @param name the named lock to get lease information for
* @return snapshot of the remote lock information
* @throws LockServiceDestroyedException if local instance of lock service has been destroyed
@@ -1892,11 +1911,11 @@ public class DLockService extends DistributedLockService {
* instance.
* <p>
* Caller must be synchronized on {@link DLockService#services}.
- *
* @see org.apache.geode.distributed.DistributedLockService#create(String, DistributedSystem)
*/
static DLockService basicCreate(String serviceName, InternalDistributedSystem ds,
- boolean isDistributed, boolean destroyOnDisconnect, boolean automateFreeResources)
+ boolean isDistributed, boolean destroyOnDisconnect,
+ boolean automateFreeResources)
throws IllegalArgumentException {
Assert.assertHoldsLock(services, true);
@@ -1911,7 +1930,9 @@ public class DLockService extends DistributedLockService {
return svc;
}
- /** initialize this DLockService object */
+ /**
+ * initialize this DLockService object
+ */
protected boolean init() {
boolean success = false;
try {
@@ -1944,7 +1965,7 @@ public class DLockService extends DistributedLockService {
* DLockService.create(Object, DistributedSystem, DistributionAdvisor)
*/
protected DLockService(String serviceName, DistributedSystem ds, boolean isDistributed,
- boolean destroyOnDisconnect, boolean automateFreeResources) {
+ boolean destroyOnDisconnect, boolean automateFreeResources) {
super();
this.dlockStats = getOrCreateStats(ds);
this.serialNumber = createSerialNumber();
@@ -2026,10 +2047,12 @@ public class DLockService extends DistributedLockService {
}
public boolean acquireTryLocks(final DLockBatch dlockBatch, final long waitTimeMillis,
- final long leaseTimeMillis, final Object[] keyIfFailed) throws InterruptedException {
+ final long leaseTimeMillis, final Object[] keyIfFailed)
+ throws InterruptedException {
checkDestroyed();
- if (Thread.interrupted())
+ if (Thread.interrupted()) {
throw new InterruptedException();
+ }
if (keyIfFailed.length < 1) {
throw new IllegalArgumentException(
LocalizedStrings.DLockService_KEYIFFAILED_MUST_HAVE_A_LENGTH_OF_ONE_OR_GREATER
@@ -2047,15 +2070,18 @@ public class DLockService extends DistributedLockService {
long requestLeaseTime = leaseTimeMillis;
// -1 means "lease forever". Long.MAX_VALUE is pretty close.
- if (requestLeaseTime == -1)
+ if (requestLeaseTime == -1) {
requestLeaseTime = Long.MAX_VALUE;
+ }
// -1 means "wait forever". Long.MAX_VALUE is pretty close.
- if (requestWaitTime == -1)
+ if (requestWaitTime == -1) {
requestWaitTime = Long.MAX_VALUE;
+ }
long waitLimit = startTime + requestWaitTime;
- if (waitLimit < 0)
+ if (waitLimit < 0) {
waitLimit = Long.MAX_VALUE;
+ }
long statStart = getStats().startLockWait();
boolean gotLocks = false;
@@ -2135,7 +2161,6 @@ public class DLockService extends DistributedLockService {
* Returns copy of the tokens map. Synchronizes on token map.
* <p>
* Called by {@link org.apache.geode.internal.admin.remote.FetchDistLockInfoResponse}.
- *
* @return copy of the tokens map
*/
public Map<Object, DLockToken> snapshotService() {
@@ -2151,7 +2176,9 @@ public class DLockService extends DistributedLockService {
return blockedOn;
}
- /** Returns true if the lock service is distributed; false if local only */
+ /**
+ * Returns true if the lock service is distributed; false if local only
+ */
public boolean isDistributed() {
return this.isDistributed;
}
@@ -2164,7 +2191,9 @@ public class DLockService extends DistributedLockService {
return this.lessorDepartureHandler;
}
- /** The name of this service */
+ /**
+ * The name of this service
+ */
public String getName() {
return this.serviceName;
}
@@ -2185,12 +2214,11 @@ public class DLockService extends DistributedLockService {
/**
* Returns true if any tokens in this service are currently held. Synchronizes on tokens map and
* on each token to check the lease.
- *
* @return true if any tokens in this service are currently held
*/
boolean hasHeldLocks() {
synchronized (this.tokens) {
- for (Iterator iter = this.tokens.values().iterator(); iter.hasNext();) {
+ for (Iterator iter = this.tokens.values().iterator(); iter.hasNext(); ) {
DLockToken token = (DLockToken) iter.next();
if (token.isLeaseHeld()) {
return true;
@@ -2216,13 +2244,15 @@ public class DLockService extends DistributedLockService {
}
}
- /** Destroys all lock services in this VM. Used in test tearDown code. */
+ /**
+ * Destroys all lock services in this VM. Used in test tearDown code.
+ */
public static void destroyAll() {
Collection svcs = Collections.EMPTY_SET;
synchronized (services) {
svcs = new HashSet(services.values());
}
- for (Iterator iter = svcs.iterator(); iter.hasNext();) {
+ for (Iterator iter = svcs.iterator(); iter.hasNext(); ) {
DLockService svc = (DLockService) iter.next();
try {
svc.destroyAndRemove();
@@ -2237,7 +2267,6 @@ public class DLockService extends DistributedLockService {
/**
* Destroys an existing service and removes it from the map
- *
* @since GemFire 3.5
*/
public void destroyAndRemove() {
@@ -2252,8 +2281,9 @@ public class DLockService extends DistributedLockService {
try {
synchronized (services) {
try {
- if (isDestroyed())
+ if (isDestroyed()) {
return;
+ }
setDestroyingThread();
synchronized (this.lockGrantorIdLock) { // force ordering in lock request
synchronized (this.destroyLock) {
@@ -2288,7 +2318,8 @@ public class DLockService extends DistributedLockService {
/**
* Unlock all locks currently held by this process, and mark it as destroyed. Returns true if
- * caller performed actual destroy. Returns false if this lock service has already been destroyed.
+ * caller performed actual destroy. Returns false if this lock service has already been
+ * destroyed.
* <p>
* {@link #services} is held for the entire operation. {@link #destroyLock} is held only while
* setting {@link #destroyed} to true and just prior to performing any real work.
@@ -2296,7 +2327,7 @@ public class DLockService extends DistributedLockService {
* Caller must be synchronized on {@link DLockService#services};
*/
private void basicDestroy(boolean isCurrentlyLockGrantor, boolean isMakingLockGrantor,
- boolean maybeHasActiveLocks) {
+ boolean maybeHasActiveLocks) {
Assert.assertHoldsLock(services, false);
// synchronized (this.serviceLock) {
final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
@@ -2318,7 +2349,7 @@ public class DLockService extends DistributedLockService {
if (!NonGrantorDestroyedProcessor.send(this.serviceName, theLockGrantorId, dm)) {
// grantor responded NOT_GRANTOR
notLockGrantorId(theLockGrantorId, 0, TimeUnit.MILLISECONDS); // nulls out grantor to
- // force call to elder
+ // force call to elder
retry = true;
}
}
@@ -2333,7 +2364,7 @@ public class DLockService extends DistributedLockService {
Assert.assertTrue(false,
LocalizedStrings.DLockService_FAILED_TO_NOTIFY_GRANTOR_OF_DESTRUCTION_WITHIN_0_ATTEMPTS
.toLocalizedString(
- new Object[] {Integer.valueOf(DEBUG_NONGRANTOR_DESTROY_LOOP_COUNT)}));
+ new Object[]{Integer.valueOf(DEBUG_NONGRANTOR_DESTROY_LOOP_COUNT)}));
}
}
@@ -2377,7 +2408,6 @@ public class DLockService extends DistributedLockService {
/**
* Called by grantor recovery to return set of locks held by this process. Synchronizes on
* lockGrantorIdLock, tokens map, and each lock token.
- *
* @param newlockGrantorId the newly recovering grantor
*/
Set<DLockRemoteToken> getLockTokensForRecovery(LockGrantorId newlockGrantorId) {
@@ -2403,7 +2433,7 @@ public class DLockService extends DistributedLockService {
synchronized (this.lockGrantorIdLock) {
synchronized (this.tokens) {
// build up set of currently held locks
- for (Iterator iter = this.tokens.values().iterator(); iter.hasNext();) {
+ for (Iterator iter = this.tokens.values().iterator(); iter.hasNext(); ) {
DLockToken token = (DLockToken) iter.next();
synchronized (token) {
if (token.isLeaseHeld()) {
@@ -2435,7 +2465,6 @@ public class DLockService extends DistributedLockService {
/**
* Returns the named lock token or null if it doesn't exist. Synchronizes on tokens map.
- *
* @return the named lock token or null if it doesn't exist
*/
public DLockToken getToken(Object name) {
@@ -2447,7 +2476,6 @@ public class DLockService extends DistributedLockService {
/**
* Returns the named lock token or null if it doesn't exist. Caller must synchronize on tokens
* map.
- *
* @return the named lock token or null if it doesn't exist
*/
private DLockToken basicGetToken(Object name) {
@@ -2457,7 +2485,6 @@ public class DLockService extends DistributedLockService {
/**
* Returns an unmodifiable collection backed by the values of the DLockToken map for testing
* purposes only. Synchronizes on tokens map.
- *
* @return an unmodifiable collection of the tokens map values
*/
public Collection getTokens() {
@@ -2467,8 +2494,8 @@ public class DLockService extends DistributedLockService {
}
/**
- * Returns an existing or creates a new DLockToken. Synchronizes on tokens map and the lock token.
- *
+ * Returns an existing or creates a new DLockToken. Synchronizes on tokens map and the lock
+ * token.
* @param name the name of the lock
* @return an existing or new instantiated lock token
*/
@@ -2501,13 +2528,12 @@ public class DLockService extends DistributedLockService {
/**
* Returns number of lock tokens currently leased by this member. Synchronizes on tokens map and
* each lock token.
- *
* @return number of lock tokens currently leased by this member
*/
private int numLocksHeldInThisVM() {
int numLocksHeld = 0;
synchronized (this.tokens) {
- for (Iterator iter = this.tokens.values().iterator(); iter.hasNext();) {
+ for (Iterator iter = this.tokens.values().iterator(); iter.hasNext(); ) {
DLockToken token = (DLockToken) iter.next();
synchronized (token) {
if (token.isLeaseHeld()) {
@@ -2528,7 +2554,7 @@ public class DLockService extends DistributedLockService {
StringBuffer buffer = new StringBuffer();
buffer.append(" ").append(this.tokens.size()).append(" tokens, ");
buffer.append(numLocksHeldInThisVM()).append(" locks held\n");
- for (Iterator iter = this.tokens.entrySet().iterator(); iter.hasNext();) {
+ for (Iterator iter = this.tokens.entrySet().iterator(); iter.hasNext(); ) {
Map.Entry entry = (Map.Entry) iter.next();
buffer.append(" ").append(entry.getKey()).append(": ");
DLockToken token = (DLockToken) entry.getValue();
@@ -2579,7 +2605,9 @@ public class DLockService extends DistributedLockService {
// System disconnect listener
// -------------------------------------------------------------------------
- /** Destroys all named locking services on disconnect from system */
+ /**
+ * Destroys all named locking services on disconnect from system
+ */
protected static final InternalDistributedSystem.DisconnectListener disconnectListener =
new InternalDistributedSystem.DisconnectListener() {
@Override
@@ -2609,24 +2637,28 @@ public class DLockService extends DistributedLockService {
// ----------------------------------------------------------------
- private static final DummyDLockStats DUMMY_STATS = new DummyDLockStats();
-
public static final SuspendLockingToken SUSPEND_LOCKING_TOKEN = new SuspendLockingToken();
// -------------------------------------------------------------------------
// Static fields
// -------------------------------------------------------------------------
- /** Map of all locking services. Key:ServiceName, Value:DLockService */
+ /**
+ * Map of all locking services. Key:ServiceName, Value:DLockService
+ */
protected static final Map<String, DLockService> services = new HashMap<String, DLockService>();
protected static final Object creationLock = new Object();
- /** All DLock threads belong to this group */
+ /**
+ * All DLock threads belong to this group
+ */
static ThreadGroup threadGroup;
- /** DLock statistics; static because multiple dlock instances can exist */
- private static DistributedLockStats stats = DUMMY_STATS;
+ /**
+ * DLock statistics; static because multiple dlock instances can exist
+ */
+ private static DistributedLockStats stats;
// -------------------------------------------------------------------------
// Reserved lock service names
@@ -2634,7 +2666,7 @@ public class DLockService extends DistributedLockService {
public static final String LTLS = "LTLS";
public static final String DTLS = "DTLS";
- static final String[] reservedNames = new String[] {LTLS, DTLS};
+ static final String[] reservedNames = new String[]{LTLS, DTLS};
// -------------------------------------------------------------------------
// DLS serial number (uniquely identifies local instance of DLS)
@@ -2649,7 +2681,6 @@ public class DLockService extends DistributedLockService {
/**
* Incrementing serial number used to identify order of DLS creation
- *
* @see DLockService#getSerialNumber()
*/
private static final AtomicInteger serialNumberSequencer = new AtomicInteger(START_SERIAL_NUMBER);
@@ -2664,7 +2695,6 @@ public class DLockService extends DistributedLockService {
* Generates a serial number for identifying a DLS. Later instances of the same named DLS will
* have a greater serial number than earlier instances. This number increments statically
* throughout the life of this JVM. Rollover to negative is allowed.
- *
* @return the new serial number
*/
protected static int createSerialNumber() {
@@ -2696,29 +2726,27 @@ public class DLockService extends DistributedLockService {
}
public static DistributedLockService create(String serviceName, InternalDistributedSystem ds,
- boolean distributed, boolean destroyOnDisconnect) {
+ boolean distributed, boolean destroyOnDisconnect) {
return create(serviceName, ds, distributed, destroyOnDisconnect, false);
}
/**
* Creates named <code>DistributedLockService</code>.
- *
* @param serviceName name of the service
* @param ds InternalDistributedSystem
* @param distributed true if lock service will be distributed; false will be local only
* @param destroyOnDisconnect true if lock service should destroy itself using system disconnect
- * listener
- * @param automateFreeResources true if freeResources should be automatically called during unlock
- *
+ * listener
+ * @param automateFreeResources true if freeResources should be automatically called during
+ * unlock
* @throws IllegalArgumentException if serviceName is invalid or this process has already created
- * named service
- *
+ * named service
* @throws IllegalStateException if system is in process of disconnecting
- *
* @see org.apache.geode.distributed.DistributedLockService#create(String, DistributedSystem)
*/
public static DistributedLockService create(String serviceName, InternalDistributedSystem ds,
- boolean distributed, boolean destroyOnDisconnect, boolean automateFreeResources)
+ boolean distributed, boolean destroyOnDisconnect,
+ boolean automateFreeResources)
throws IllegalArgumentException, IllegalStateException {
// basicCreate will construct DLockService and it calls getOrCreateStats...
synchronized (creationLock) {
@@ -2780,15 +2808,14 @@ public class DLockService extends DistributedLockService {
/**
* Fills lists of service names.
- *
* @param grantors filled with service names of all services we are currently the grantor of.
* @param grantorVersions elder assigned grantor version ids
* @param grantorSerialNumbers member specific DLS serial number hosting grantor
* @param nonGrantors filled with service names of all services we have that we are not the
- * grantor of.
+ * grantor of.
*/
public static void recoverRmtElder(ArrayList grantors, ArrayList grantorVersions,
- ArrayList grantorSerialNumbers, ArrayList nonGrantors) {
+ ArrayList grantorSerialNumbers, ArrayList nonGrantors) {
synchronized (services) {
Iterator entries = services.entrySet().iterator();
while (entries.hasNext()) {
@@ -2814,7 +2841,6 @@ public class DLockService extends DistributedLockService {
* Called when an elder is doing recovery. For every service that we are the grantor for add it to
* the grantorMap For every service we have that is not in the grantorMap add its name to
* needsRecovery set.
- *
* @param dm our local DM
*/
public static void recoverLocalElder(DistributionManager dm, Map grantors, Set needsRecovery) {
@@ -2851,12 +2877,16 @@ public class DLockService extends DistributedLockService {
// Public static methods
// -------------------------------------------------------------------------
- /** Convenience method to get named DLockService */
+ /**
+ * Convenience method to get named DLockService
+ */
public static DLockService getInternalServiceNamed(String serviceName) {
return (DLockService) services.get(serviceName);
}
- /** Validates service name for external creation */
+ /**
+ * Validates service name for external creation
+ */
public static void validateServiceName(String serviceName) {
if (serviceName == null || serviceName.length() == 0) {
throw new IllegalArgumentException(
@@ -2872,7 +2902,9 @@ public class DLockService extends DistributedLockService {
}
}
- /** Return a snapshot of all services */
+ /**
+ * Return a snapshot of all services
+ */
public static Map<String, DLockService> snapshotAllServices() { // used by: internal/admin/remote
Map snapshot = null;
synchronized (services) {
@@ -2914,21 +2946,19 @@ public class DLockService extends DistributedLockService {
/**
* Return a timestamp that represents the current time for locking purposes.
- *
* @since GemFire 3.5
*/
static long getLockTimeStamp(DistributionManager dm) {
return dm.cacheTimeMillis();
}
- /** Get or create static dlock stats */
- protected static synchronized DistributedLockStats getOrCreateStats(DistributedSystem distributedSystem) {
- if (stats == DUMMY_STATS) {
- Assert.assertTrue(distributedSystem != null, "Need an instance of InternalDistributedSystem");
- StatisticsFactory statFactory = distributedSystem.getStatisticsFactory();
- long statId = OSProcess.getId();
- stats = new DLockStats(statFactory, statId);
- }
+ /**
+ * Get or create static dlock stats
+ */
+ protected static synchronized DistributedLockStats getOrCreateStats(
+ DistributedSystem distributedSystem) {
+ long statId = OSProcess.getId();
+ stats = new DLockStats(statId);
return stats;
}
@@ -2975,14 +3005,12 @@ public class DLockService extends DistributedLockService {
}
static void closeStats() {
- if (stats != DUMMY_STATS) {
- ((DLockStats) stats).close();
- stats = DUMMY_STATS;
- }
threadGroup = null;
}
- /** Provide way to peek at current lock grantor id when dls does not exist */
+ /**
+ * Provide way to peek at current lock grantor id when dls does not exist
+ */
static GrantorInfo checkLockGrantorInfo(String serviceName, InternalDistributedSystem system) {
GrantorInfo gi = GrantorRequestProcessor.peekGrantor(serviceName, system);
if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
@@ -3009,14 +3037,18 @@ public class DLockService extends DistributedLockService {
// SuspendLockingToken inner class
// -------------------------------------------------------------------------
- /** Used as the name (key) for the suspend locking entry in the tokens map */
+ /**
+ * Used as the name (key) for the suspend locking entry in the tokens map
+ */
public static class SuspendLockingToken implements DataSerializableFixedID {
- public SuspendLockingToken() {}
+ public SuspendLockingToken() {
+ }
@Override
public boolean equals(Object o) {
- if (o == null)
+ if (o == null) {
return false;
+ }
return o instanceof SuspendLockingToken;
}
@@ -3030,9 +3062,11 @@ public class DLockService extends DistributedLockService {
return SUSPEND_LOCKING_TOKEN;
}
- public void fromData(DataInput in) {}
+ public void fromData(DataInput in) {
+ }
- public void toData(DataOutput out) {}
+ public void toData(DataOutput out) {
+ }
@Override
public Version[] getSerializationVersions() {
@@ -3053,7 +3087,6 @@ public class DLockService extends DistributedLockService {
/**
* Creates a new DLockStopper for the specified DLockService and DM.
- *
* @param dm the DM to check for shutdown
* @param dls the DLockService to check for DLS destroy
*/
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/MembershipManager.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/MembershipManager.java
index 7c227ed..9609f42 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/MembershipManager.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/MembershipManager.java
@@ -22,8 +22,8 @@ import java.util.concurrent.locks.ReadWriteLock;
import org.apache.geode.SystemFailure;
import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.distributed.internal.DMStats;
import org.apache.geode.distributed.internal.DistributionMessage;
+import org.apache.geode.statistics.distributed.DMStats;
/**
* A MembershipManager is responsible for reporting a MemberView, as well as having explicit
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketAdvisor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketAdvisor.java
index 47f4a82..ee603a1 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketAdvisor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketAdvisor.java
@@ -66,6 +66,7 @@ import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.util.StopWatch;
+import org.apache.geode.statistics.region.PartitionedRegionStats;
/**
* Specialized {@link CacheDistributionAdvisor} for {@link BucketRegion BucketRegions}. The
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
index 0fb745d..d92f65d 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
@@ -55,7 +55,6 @@ import org.apache.geode.distributed.internal.AtomicLongWithTerminalState;
import org.apache.geode.distributed.internal.DirectReplyProcessor;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.DistributionConfig;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.HeapDataOutputStream;
@@ -94,6 +93,7 @@ import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.offheap.annotations.Unretained;
+import org.apache.geode.statistics.cache.CachePerfStats;
/**
@@ -1510,7 +1510,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
checkReadiness();
checkForNoAccess();
CachePerfStats stats = getCachePerfStats();
- long start = stats.startGet();
+ long start = System.nanoTime();
boolean miss = true;
try {
@@ -1824,7 +1824,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
&& ((org.apache.geode.Delta) instance).hasDelta()) {
try {
HeapDataOutputStream hdos = new HeapDataOutputStream(Version.CURRENT);
- long start = DistributionStats.getStatTime();
+ long start = System.nanoTime();
((org.apache.geode.Delta) instance).toDelta(hdos);
event.setDeltaBytes(hdos.toByteArray());
this.partitionedRegion.getCachePerfStats().endDeltaPrepared(start);
@@ -1914,8 +1914,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
// routing tables for any of the receivers
Set recipients = null;
Set membersWithRouting = filterRoutingInfo.getMembers();
- for (Iterator it = membersWithRouting.iterator(); it.hasNext();) {
- Object mbr = it.next();
+ for (Object mbr : membersWithRouting) {
if (!cacheOpRecipients.contains(mbr)) {
// anyWithRouting = true;
if (!adjunctRecipients.contains(mbr)) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java
index 7d63a3d..22c888a 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java
@@ -95,7 +95,6 @@ import org.apache.geode.internal.cache.eviction.EvictableEntry;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionExecutor;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultWaiter;
-import org.apache.geode.internal.cache.execute.FunctionStats;
import org.apache.geode.internal.cache.execute.LocalResultCollector;
import org.apache.geode.internal.cache.execute.RegionFunctionContextImpl;
import org.apache.geode.internal.cache.execute.ServerToClientFunctionResultSender;
@@ -127,6 +126,7 @@ import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.sequencelog.RegionLogger;
import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
import org.apache.geode.statistics.disk.DiskRegionStats;
+import org.apache.geode.statistics.function.FunctionStats;
@SuppressWarnings("deprecation")
public class DistributedRegion extends LocalRegion implements InternalDistributedRegion {
@@ -2268,7 +2268,7 @@ public class DistributedRegion extends LocalRegion implements InternalDistribute
// Set eventId. Required for interested clients.
event.setNewEventId(cache.getDistributedSystem());
- long startPut = CachePerfStats.getStatTime();
+ long startPut = System.nanoTime();
validateKey(event.getKey());
// this next step also distributes the object to other processes, if necessary
try {
@@ -3702,9 +3702,9 @@ public class DistributedRegion extends LocalRegion implements InternalDistribute
ResultSender resultSender = new DistributedRegionFunctionResultSender(dm, msg, function);
final RegionFunctionContextImpl context = new RegionFunctionContextImpl(cache, function.getId(),
this, args, filter, null, null, resultSender, isReExecute);
- FunctionStats stats = FunctionStats.getFunctionStats(function.getId(), dm.getSystem());
+ FunctionStats stats = FunctionStats.getFunctionStats(function.getId());
try {
- long start = stats.startTime();
+ long start = System.nanoTime();
stats.startFunctionExecution(function.hasResult());
function.execute(context);
stats.endFunctionExecution(start, function.hasResult());
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCacheForClientAccess.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCacheForClientAccess.java
index 8062111..e6e4c39 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCacheForClientAccess.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCacheForClientAccess.java
@@ -91,6 +91,7 @@ import org.apache.geode.pdx.PdxInstanceFactory;
import org.apache.geode.pdx.PdxSerializer;
import org.apache.geode.pdx.internal.TypeRegistry;
import org.apache.geode.security.NotAuthorizedException;
+import org.apache.geode.statistics.cache.CachePerfStats;
/**
* This class delegates all methods to the InternalCache instance
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
index e5a13bb..5f0f72a 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
@@ -53,7 +53,6 @@ import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireException;
-import org.apache.geode.statistics.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
@@ -251,7 +250,9 @@ import org.apache.geode.internal.sequencelog.RegionLogger;
import org.apache.geode.internal.size.Sizeable;
import org.apache.geode.internal.util.TransformUtils;
import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
+import org.apache.geode.statistics.cache.CachePerfStats;
import org.apache.geode.statistics.disk.DiskRegionStats;
+import org.apache.geode.statistics.region.PartitionedRegionStats;
/**
* A Region whose total storage is split into chunks of data (partitions) which are copied up to a
@@ -541,7 +542,7 @@ public class PartitionedRegion extends LocalRegion
@Override
public boolean remove(Object key, Object value, Object callbackArg) {
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
try {
return super.remove(key, value, callbackArg);
} finally {
@@ -741,7 +742,7 @@ public class PartitionedRegion extends LocalRegion
super(regionName, regionAttributes, parentRegion, cache, internalRegionArgs);
this.node = initializeNode();
- this.prStats = new PartitionedRegionStats(cache.getDistributedSystem().getStatisticsFactory(), getFullPath());
+ this.prStats = new PartitionedRegionStats(getFullPath());
this.regionIdentifier = getFullPath().replace('/', '#');
if (logger.isDebugEnabled()) {
@@ -832,8 +833,7 @@ public class PartitionedRegion extends LocalRegion
&& !this.getEvictionAttributes().getAlgorithm().isNone()
&& this.getEvictionAttributes().getAction().isOverflowToDisk())
|| this.getDataPolicy().withPersistence()) {
- StatisticsFactory sf = this.getCache().getDistributedSystem().getStatisticsFactory();
- this.diskRegionStats = new DiskRegionStats(sf, getFullPath());
+ this.diskRegionStats = new DiskRegionStats(getFullPath());
} else {
this.diskRegionStats = null;
}
@@ -1618,7 +1618,7 @@ public class PartitionedRegion extends LocalRegion
@Override
protected Region.Entry<?, ?> nonTXGetEntry(KeyInfo keyInfo, boolean access,
boolean allowTombstones) {
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
final Object key = keyInfo.getKey();
try {
int bucketId = keyInfo.getBucketId();
@@ -2051,7 +2051,7 @@ public class PartitionedRegion extends LocalRegion
public boolean virtualPut(EntryEventImpl event, boolean ifNew, boolean ifOld,
Object expectedOldValue, boolean requireOldValue, long lastModified,
boolean overwriteDestroyed) throws TimeoutException, CacheWriterException {
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
boolean result = false;
final DistributedPutAllOperation putAllOp_save = event.setPutAllOperation(null);
@@ -2224,7 +2224,7 @@ public class PartitionedRegion extends LocalRegion
throw cache.getCacheClosedException("Cache is shutting down");
}
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
// build all the msgs by bucketid
HashMap prMsgMap = putAllOp.createPRMessages();
PutAllPartialResult partialKeys = new PutAllPartialResult(putAllOp.putAllDataSize);
@@ -2316,7 +2316,7 @@ public class PartitionedRegion extends LocalRegion
throw cache.getCacheClosedException("Cache is shutting down");
}
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
// build all the msgs by bucketid
HashMap<Integer, RemoveAllPRMessage> prMsgMap = op.createPRMessages();
PutAllPartialResult partialKeys = new PutAllPartialResult(op.removeAllDataSize);
@@ -3327,7 +3327,7 @@ public class PartitionedRegion extends LocalRegion
}
// Potentially no storage assigned, start bucket creation, be careful of race
// conditions
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
if (isDataStore()) {
ret = this.redundancyProvider.createBucketAtomically(bucketId, size, startTime, false,
partitionName);
@@ -3345,7 +3345,7 @@ public class PartitionedRegion extends LocalRegion
Object obj = null;
final Object key = keyInfo.getKey();
final Object aCallbackArgument = keyInfo.getCallbackArg();
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
try {
int bucketId = keyInfo.getBucketId();
if (bucketId == KeyInfo.UNKNOWN_BUCKET) {
@@ -5141,7 +5141,7 @@ public class PartitionedRegion extends LocalRegion
final Object expectedOldValue)
throws TimeoutException, EntryNotFoundException, CacheWriterException {
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
try {
if (event.getEventId() == null) {
event.setNewEventId(this.cache.getDistributedSystem());
@@ -5551,7 +5551,6 @@ public class PartitionedRegion extends LocalRegion
}
}
this.distAdvisor.close();
- getPrStats().close();
if (getDiskStore() != null && getDiskStore().getOwnedByRegion()) {
getDiskStore().close();
}
@@ -5631,7 +5630,7 @@ public class PartitionedRegion extends LocalRegion
@Override
public void basicInvalidate(EntryEventImpl event) throws EntryNotFoundException {
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
try {
if (event.getEventId() == null) {
event.setNewEventId(this.cache.getDistributedSystem());
@@ -6277,7 +6276,7 @@ public class PartitionedRegion extends LocalRegion
@Override
protected boolean nonTXContainsKey(KeyInfo keyInfo) {
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
boolean contains = false;
try {
int bucketId = keyInfo.getBucketId();
@@ -6451,7 +6450,7 @@ public class PartitionedRegion extends LocalRegion
// checkClosed();
checkReadiness();
validateKey(key);
- final long startTime = PartitionedRegionStats.startTime();
+ final long startTime = System.nanoTime();
boolean containsValueForKey = false;
try {
containsValueForKey = getDataView().containsValueForKey(getKeyInfo(key), this);
@@ -7535,7 +7534,6 @@ public class PartitionedRegion extends LocalRegion
}
} finally {
this.getRegionAdvisor().close();
- getPrStats().close();
this.cache.getInternalResourceManager(false).removeResourceListener(this);
this.locallyDestroyingThread = null;
if (logger.isDebugEnabled()) {
@@ -7633,7 +7631,6 @@ public class PartitionedRegion extends LocalRegion
} finally {
// tell other members to recover redundancy for any buckets
this.getRegionAdvisor().close();
- getPrStats().close();
}
} else if (Operation.REGION_DESTROY.equals(op) || Operation.REGION_EXPIRE_DESTROY.equals(op)) {
if (logger.isDebugEnabled()) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java
index 9048851..97a85d8 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java
@@ -74,12 +74,10 @@ import org.apache.geode.distributed.internal.membership.InternalDistributedMembe
import org.apache.geode.i18n.StringId;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.BucketRegion.RawValue;
-import org.apache.geode.internal.cache.LocalRegion.RegionPerfStats;
import org.apache.geode.internal.cache.PartitionedRegion.BucketLock;
import org.apache.geode.internal.cache.PartitionedRegion.SizeEntry;
import org.apache.geode.internal.cache.backup.BackupService;
import org.apache.geode.internal.cache.execute.BucketMovedException;
-import org.apache.geode.internal.cache.execute.FunctionStats;
import org.apache.geode.internal.cache.execute.PartitionedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.RegionFunctionContextImpl;
import org.apache.geode.internal.cache.partitioned.Bucket;
@@ -102,6 +100,9 @@ import org.apache.geode.internal.logging.log4j.LocalizedMessage;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock.StoppableReadLock;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock.StoppableWriteLock;
+import org.apache.geode.statistics.cache.CachePerfStats;
+import org.apache.geode.statistics.function.FunctionStats;
+import org.apache.geode.statistics.region.RegionPerfStats;
/**
* Implementation of DataStore (DS) for a PartitionedRegion (PR). This will be import
@@ -206,8 +207,7 @@ public class PartitionedRegionDataStore implements HasCachePerfStats {
this.maximumLocalBytes = (pr.getLocalMaxMemory() * PartitionedRegionHelper.BYTES_PER_MB);
// this.bucketStats = new CachePerfStats(pr.getSystem(), "partition-" + pr.getName());
- this.bucketStats =
- new RegionPerfStats(pr.getCache(), pr.getCachePerfStats(), "partition-" + pr.getName());
+ this.bucketStats = new RegionPerfStats(pr.getCachePerfStats(), "partition-" + pr.getName());
this.keysOfInterest = new ConcurrentHashMap();
}
@@ -1472,7 +1472,6 @@ public class PartitionedRegionDataStore implements HasCachePerfStats {
this.partitionedRegion.getFullPath()), ex);
} finally {
this.partitionedRegion.getPrStats().setBucketCount(0);
- this.bucketStats.close();
}
}
@@ -3009,9 +3008,9 @@ public class PartitionedRegionDataStore implements HasCachePerfStats {
.constructAndGetAllColocatedLocalDataSet(this.partitionedRegion, bucketSet),
bucketSet, resultSender, isReExecute);
- FunctionStats stats = FunctionStats.getFunctionStats(function.getId(), dm.getSystem());
+ FunctionStats stats = FunctionStats.getFunctionStats(function.getId());
try {
- long start = stats.startTime();
+ long start = System.nanoTime();
stats.startFunctionExecution(function.hasResult());
if (logger.isDebugEnabled()) {
logger.debug("Executing Function: {} on Remote Node with context: ", function.getId(),
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionRedundancyTracker.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionRedundancyTracker.java
index 54941bb..eee997d 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionRedundancyTracker.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionRedundancyTracker.java
@@ -20,6 +20,7 @@ import org.apache.logging.log4j.Logger;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
+import org.apache.geode.statistics.region.PartitionedRegionStats;
/**
* Keeps track redundancy statistics across the buckets of a given {@link PartitionedRegion}
@@ -45,7 +46,7 @@ class PartitionedRegionRedundancyTracker {
* @param regionPath the full path of the region to track
*/
PartitionedRegionRedundancyTracker(int totalBuckets, int redundantCopies,
- PartitionedRegionStats stats, String regionPath) {
+ PartitionedRegionStats stats, String regionPath) {
this.stats = stats;
this.regionPath = regionPath;
this.totalBuckets = totalBuckets;
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/ContainsKeyValueMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/ContainsKeyValueMessage.java
index 3d139e4..70b8af5 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/ContainsKeyValueMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/ContainsKeyValueMessage.java
@@ -28,7 +28,6 @@ import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DirectReplyProcessor;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
@@ -215,7 +214,7 @@ public class ContainsKeyValueMessage extends PartitionMessageWithDirectReply {
if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
logger.trace(LogMarker.DM_VERBOSE, "{} Processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/DumpB2NRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/DumpB2NRegion.java
index 0a90eb7..6a48011 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/DumpB2NRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/DumpB2NRegion.java
@@ -34,7 +34,6 @@ import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
@@ -207,7 +206,7 @@ public class DumpB2NRegion extends PartitionMessage {
if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
logger.trace(LogMarker.DM_VERBOSE, "{} processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
@@ -329,7 +328,6 @@ public class DumpB2NRegion extends PartitionMessage {
*/
@Override
protected Object clone() throws CloneNotSupportedException {
- // TODO Auto-generated method stub
return super.clone();
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchBulkEntriesMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchBulkEntriesMessage.java
index 0916e37..5823c2a 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchBulkEntriesMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchBulkEntriesMessage.java
@@ -36,7 +36,6 @@ import org.apache.geode.DataSerializer;
import org.apache.geode.cache.CacheException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
@@ -421,7 +420,7 @@ public class FetchBulkEntriesMessage extends PartitionMessage {
logger.trace(LogMarker.DM_VERBOSE, "{} processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchEntriesMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchEntriesMessage.java
index d54186c..a2cd032 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchEntriesMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchEntriesMessage.java
@@ -34,7 +34,6 @@ import org.apache.geode.cache.CacheException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
@@ -370,7 +369,7 @@ public class FetchEntriesMessage extends PartitionMessage {
logger.trace(LogMarker.DM_VERBOSE, "{} processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchKeysMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchKeysMessage.java
index a62b1b4..9f773e9 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchKeysMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchKeysMessage.java
@@ -31,7 +31,6 @@ import org.apache.geode.DataSerializer;
import org.apache.geode.cache.CacheException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
@@ -428,7 +427,7 @@ public class FetchKeysMessage extends PartitionMessage {
logger.trace(LogMarker.DM_VERBOSE, "{} processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/GetMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/GetMessage.java
index 6a419cb..6440d76 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/GetMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/GetMessage.java
@@ -31,7 +31,6 @@ import org.apache.geode.distributed.internal.DirectReplyProcessor;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
@@ -64,13 +63,11 @@ import org.apache.geode.internal.offheap.OffHeapHelper;
import org.apache.geode.internal.util.BlobHelper;
/**
- * This message is used as the request for a
- * {@link org.apache.geode.cache.Region#get(Object)}operation. The reply is sent in a
- * {@link org.apache.geode.internal.cache.partitioned.GetMessage}.
+ * This message is used as the request for a {@link org.apache.geode.cache.Region#get(Object)}operation.
+ * The reply is sent in a {@link org.apache.geode.internal.cache.partitioned.GetMessage}.
*
* Since the {@link org.apache.geode.cache.Region#get(Object)}operation is used <bold>very </bold>
* frequently the performance of this class is critical.
- *
* @since GemFire 5.0
*/
public class GetMessage extends PartitionMessageWithDirectReply {
@@ -78,10 +75,14 @@ public class GetMessage extends PartitionMessageWithDirectReply {
private Object key;
- /** The callback arg of the operation */
+ /**
+ * The callback arg of the operation
+ */
private Object cbArg;
- /** set to true if the message was requeued in the pr executor pool */
+ /**
+ * set to true if the message was requeued in the pr executor pool
+ */
private transient boolean forceUseOfPRExecutor;
private ClientProxyMembershipID context;
@@ -95,11 +96,13 @@ public class GetMessage extends PartitionMessageWithDirectReply {
/**
* Empty constructor to satisfy {@link DataSerializer} requirements
*/
- public GetMessage() {}
+ public GetMessage() {
+ }
private GetMessage(InternalDistributedMember recipient, int regionId,
- DirectReplyProcessor processor, final Object key, final Object aCallbackArgument,
- ClientProxyMembershipID context, boolean returnTombstones) {
+ DirectReplyProcessor processor, final Object key,
+ final Object aCallbackArgument,
+ ClientProxyMembershipID context, boolean returnTombstones) {
super(recipient, regionId, processor);
this.key = key;
this.cbArg = aCallbackArgument;
@@ -155,7 +158,8 @@ public class GetMessage extends PartitionMessageWithDirectReply {
@Override
protected boolean operateOnPartitionedRegion(final ClusterDistributionManager dm,
- PartitionedRegion r, long startTime) throws ForceReattemptException {
+ PartitionedRegion r, long startTime)
+ throws ForceReattemptException {
if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
logger.trace(LogMarker.DM_VERBOSE, "GetMessage operateOnRegion: {}", r.getFullPath());
}
@@ -270,7 +274,6 @@ public class GetMessage extends PartitionMessageWithDirectReply {
/**
* Sends a PartitionedRegion {@link org.apache.geode.cache.Region#get(Object)} message
- *
* @param recipient the member that the get message is sent to
* @param r the PartitionedRegion for which get was performed upon
* @param key the object to which the value should be feteched
@@ -279,8 +282,9 @@ public class GetMessage extends PartitionMessageWithDirectReply {
* @throws ForceReattemptException if the peer is no longer available
*/
public static GetResponse send(InternalDistributedMember recipient, PartitionedRegion r,
- final Object key, final Object aCallbackArgument, ClientProxyMembershipID requestingClient,
- boolean returnTombstones) throws ForceReattemptException {
+ final Object key, final Object aCallbackArgument,
+ ClientProxyMembershipID requestingClient,
+ boolean returnTombstones) throws ForceReattemptException {
Assert.assertTrue(recipient != null, "PRDistribuedGetReplyMessage NULL reply message");
GetResponse p = new GetResponse(r.getSystem(), Collections.singleton(recipient), key);
GetMessage m = new GetMessage(recipient, r.getPRId(), p, key, aCallbackArgument,
@@ -296,13 +300,11 @@ public class GetMessage extends PartitionMessageWithDirectReply {
}
/**
- * This message is used for the reply to a
- * {@link org.apache.geode.cache.Region#get(Object)}operation This is the reply to a
- * {@link GetMessage}.
+ * This message is used for the reply to a {@link org.apache.geode.cache.Region#get(Object)}operation
+ * This is the reply to a {@link GetMessage}.
*
* Since the {@link org.apache.geode.cache.Region#get(Object)}operation is used <bold>very </bold>
* frequently the performance of this class is critical.
- *
* @since GemFire 5.0
*/
public static class GetReplyMessage extends ReplyMessage {
@@ -342,7 +344,8 @@ public class GetMessage extends PartitionMessageWithDirectReply {
/**
* Empty constructor to conform to DataSerializable interface
*/
- public GetReplyMessage() {}
+ public GetReplyMessage() {
+ }
private GetReplyMessage(int processorId, RawValue val, VersionTag versionTag) {
setProcessorId(processorId);
@@ -360,7 +363,9 @@ public class GetMessage extends PartitionMessageWithDirectReply {
this.versionTag = versionTag;
}
- /** GetReplyMessages are always processed in-line */
+ /**
+ * GetReplyMessages are always processed in-line
+ */
@Override
public boolean getInlineProcess() {
return true;
@@ -370,7 +375,6 @@ public class GetMessage extends PartitionMessageWithDirectReply {
* Return the value from the get operation, serialize it bytes as late as possible to avoid
* making un-neccesary byte[] copies. De-serialize those same bytes as late as possible to avoid
* using precious threads (aka P2P readers).
- *
* @param recipient the origin VM that performed the get
* @param processorId the processor on which the origin thread is waiting
* @param val the raw value that will eventually be serialized
@@ -378,7 +382,8 @@ public class GetMessage extends PartitionMessageWithDirectReply {
* @param versionTag the version of the object
*/
public static void send(InternalDistributedMember recipient, int processorId, RawValue val,
- ReplySender replySender, VersionTag versionTag) throws ForceReattemptException {
+ ReplySender replySender, VersionTag versionTag)
+ throws ForceReattemptException {
Assert.assertTrue(recipient != null, "PRDistribuedGetReplyMessage NULL reply message");
GetReplyMessage m = new GetReplyMessage(processorId, val, versionTag);
m.setRecipient(recipient);
@@ -387,7 +392,6 @@ public class GetMessage extends PartitionMessageWithDirectReply {
/**
* Processes this message. This method is invoked by the receiver of the message.
- *
* @param dm the distribution manager that is processing the message.
*/
@Override
@@ -416,7 +420,7 @@ public class GetMessage extends PartitionMessageWithDirectReply {
if (isDebugEnabled) {
logger.debug("{} Processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
@Override
@@ -485,9 +489,7 @@ public class GetMessage extends PartitionMessageWithDirectReply {
}
/**
- * A processor to capture the value returned by
- * {@link org.apache.geode.internal.cache.partitioned.GetMessage.GetReplyMessage}
- *
+ * A processor to capture the value returned by {@link org.apache.geode.internal.cache.partitioned.GetMessage.GetReplyMessage}
* @since GemFire 5.0
*/
public static class GetResponse extends PartitionResponse {
@@ -504,9 +506,7 @@ public class GetMessage extends PartitionMessageWithDirectReply {
@Override
public void process(DistributionMessage msg) {
- if (DistributionStats.enableClockStats) {
- this.start = DistributionStats.getStatTime();
- }
+ this.start = System.nanoTime();
if (msg instanceof GetReplyMessage) {
GetReplyMessage reply = (GetReplyMessage) msg;
// De-serialization needs to occur in the requesting thread, not a P2P thread
@@ -524,9 +524,8 @@ public class GetMessage extends PartitionMessageWithDirectReply {
/**
* De-seralize the value, if the value isn't already a byte array, this method should be called
* in the context of the requesting thread for the best scalability
- *
- * @see EntryEventImpl#deserialize(byte[])
* @return the value object
+ * @see EntryEventImpl#deserialize(byte[])
*/
public Object getValue(boolean preferCD) throws ForceReattemptException {
final GetReplyMessage reply = this.getReply;
@@ -580,9 +579,7 @@ public class GetMessage extends PartitionMessageWithDirectReply {
public Object waitForResponse(boolean preferCD) throws ForceReattemptException {
try {
waitForCacheException();
- if (DistributionStats.enableClockStats) {
- getDistributionManager().getStats().incReplyHandOffTime(this.start);
- }
+ getDistributionManager().getStats().incReplyHandOffTime(this.start);
} catch (ForceReattemptException e) {
e.checkKey(key);
final String msg = "GetResponse got ForceReattemptException; rethrowing";
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/IdentityRequestMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/IdentityRequestMessage.java
index 262cc86..53f2c79 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/IdentityRequestMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/IdentityRequestMessage.java
@@ -26,7 +26,6 @@ import org.apache.geode.SystemFailure;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.MessageWithReply;
@@ -211,7 +210,7 @@ public class IdentityRequestMessage extends DistributionMessage implements Messa
if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
logger.trace(LogMarker.DM_VERBOSE, "{} Processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/InterestEventMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/InterestEventMessage.java
index 985c861..c92cb79 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/InterestEventMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/InterestEventMessage.java
@@ -27,7 +27,6 @@ import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.InterestRegistrationEvent;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyException;
@@ -192,7 +191,7 @@ public class InterestEventMessage extends PartitionMessage {
logger.trace("{} processed {}", processor, this);
}
} finally {
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/ManageBucketMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/ManageBucketMessage.java
index 8e4098f..e9e919c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/ManageBucketMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/ManageBucketMessage.java
@@ -27,7 +27,6 @@ import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
@@ -303,7 +302,7 @@ public class ManageBucketMessage extends PartitionMessage {
if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
logger.trace(LogMarker.DM_VERBOSE, "{} processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RegionAdvisor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RegionAdvisor.java
index c6a9c30..4661413 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RegionAdvisor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/RegionAdvisor.java
@@ -61,7 +61,6 @@ import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.Node;
import org.apache.geode.internal.cache.PRHARedundancyProvider.DataStoreBuckets;
import org.apache.geode.internal.cache.PartitionedRegion;
-import org.apache.geode.internal.cache.PartitionedRegionStats;
import org.apache.geode.internal.cache.ProxyBucketRegion;
import org.apache.geode.internal.cache.control.MemoryThresholds;
import org.apache.geode.internal.cache.control.ResourceAdvisor;
@@ -71,6 +70,7 @@ import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
import org.apache.geode.internal.logging.log4j.LogMarker;
+import org.apache.geode.statistics.region.PartitionedRegionStats;
public class RegionAdvisor extends CacheDistributionAdvisor {
private static final Logger logger = LogService.getLogger();
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/SizeMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/SizeMessage.java
index 55a9709..573fc97 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/SizeMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/SizeMessage.java
@@ -29,7 +29,6 @@ import org.apache.geode.cache.CacheException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
@@ -255,7 +254,7 @@ public class SizeMessage extends PartitionMessage {
if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
logger.trace(LogMarker.DM_VERBOSE, "{} processed {}", processor, this);
}
- dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
+ dm.getStats().incReplyMessageTime(System.nanoTime() - startTime);
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PersistenceAdvisorImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PersistenceAdvisorImpl.java
index 3351edd..e096b2b 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PersistenceAdvisorImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PersistenceAdvisorImpl.java
@@ -42,7 +42,6 @@ import org.apache.geode.internal.CopyOnWriteHashSet;
import org.apache.geode.internal.cache.CacheDistributionAdvisor;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.InitialImageAdvice;
-import org.apache.geode.internal.cache.DiskRegionStats;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager.MemberRevocationListener;
import org.apache.geode.internal.cache.persistence.PersistentStateQueryMessage.PersistentStateQueryReplyProcessor;
import org.apache.geode.internal.i18n.LocalizedStrings;
@@ -51,6 +50,7 @@ import org.apache.geode.internal.logging.log4j.LocalizedMessage;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.process.StartupStatus;
import org.apache.geode.internal.util.TransformUtils;
+import org.apache.geode.statistics.disk.DiskRegionStats;
public class PersistenceAdvisorImpl implements InternalPersistenceAdvisor {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
index f94915a..8f804ac 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
@@ -86,7 +86,6 @@ import org.apache.geode.internal.cache.partitioned.AllBucketProfilesUpdateMessag
import org.apache.geode.internal.cache.tier.Acceptor;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.CommunicationMode;
-import org.apache.geode.internal.cache.wan.GatewayReceiverStats;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.LoggingThreadGroup;
@@ -99,6 +98,7 @@ import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.tcp.ConnectionTable;
import org.apache.geode.internal.util.ArrayUtils;
import org.apache.geode.statistics.cache.CacheServerStats;
+import org.apache.geode.statistics.wan.GatewayReceiverStats;
/**
* Implements the acceptor thread on the bridge server. Accepts connections from the edge and starts
@@ -530,7 +530,7 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
LocalizedStrings.AcceptorImpl_CACHE_SERVER_CONNECTION_LISTENER_BOUND_TO_ADDRESS_0_WITH_BACKLOG_1,
new Object[] {sockName, Integer.valueOf(backLog)}));
if (isGatewayReceiver) {
- this.stats = GatewayReceiverStats.createGatewayReceiverStats(sockName);
+ this.stats = new GatewayReceiverStats(sockName);
} else {
this.stats = new CacheServerStats(sockName);
}
@@ -592,7 +592,6 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
return new ThreadPoolExecutor(1, HANDSHAKE_POOL_SIZE, 60, TimeUnit.SECONDS, blockingQueue,
socketThreadFactory, rejectedExecutionHandler);
} catch (IllegalArgumentException poolInitException) {
- this.stats.close();
this.serverSock.close();
this.pool.shutdown();
throw poolInitException;
@@ -664,7 +663,6 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
TimeUnit.MILLISECONDS, new SynchronousQueue(), socketThreadFactory);
}
} catch (IllegalArgumentException poolInitException) {
- this.stats.close();
this.serverSock.close();
throw poolInitException;
}
@@ -1221,9 +1219,6 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
}
} catch (IOException ignore) {
}
- if (this.stats != null) {
- this.stats.close();
- }
}
}
@@ -1629,7 +1624,6 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
shutdownSCs();
this.clientNotifier.shutdown(this.acceptorId);
shutdownPools();
- this.stats.close();
notifyCacheMembersOfClose();
} // synchronized
} catch (RuntimeException e) {/* ignore and log */
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/OriginalServerConnection.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/OriginalServerConnection.java
index f21f701..ff85136 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/OriginalServerConnection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/OriginalServerConnection.java
@@ -24,6 +24,7 @@ import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
import org.apache.geode.internal.security.SecurityService;
+import org.apache.geode.statistics.cache.CacheServerStats;
/**
* Handles everything but the new client protocol.
@@ -42,9 +43,9 @@ public class OriginalServerConnection extends ServerConnection {
*
*/
public OriginalServerConnection(Socket socket, InternalCache internalCache,
- CachedRegionHelper helper, CacheServerStats stats, int hsTimeout, int socketBufferSize,
- String communicationModeStr, byte communicationMode, Acceptor acceptor,
- SecurityService securityService) {
+ CachedRegionHelper helper, CacheServerStats stats, int hsTimeout, int socketBufferSize,
+ String communicationModeStr, byte communicationMode, Acceptor acceptor,
+ SecurityService securityService) {
super(socket, internalCache, helper, stats, hsTimeout, socketBufferSize, communicationModeStr,
communicationMode, acceptor, securityService);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ProtobufServerConnection.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ProtobufServerConnection.java
index 8392eaf..4a2894c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ProtobufServerConnection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ProtobufServerConnection.java
@@ -32,6 +32,7 @@ import org.apache.geode.internal.cache.tier.Acceptor;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.CommunicationMode;
import org.apache.geode.internal.security.SecurityService;
+import org.apache.geode.statistics.cache.CacheServerStats;
/**
* Holds the socket and protocol handler for the new client protocol.
@@ -48,9 +49,9 @@ public class ProtobufServerConnection extends ServerConnection {
* edge client over a given <code>Socket</code>.
*/
public ProtobufServerConnection(Socket socket, InternalCache c, CachedRegionHelper helper,
- CacheServerStats stats, int hsTimeout, int socketBufferSize, String communicationModeStr,
- byte communicationMode, Acceptor acceptor, ClientProtocolProcessor clientProtocolProcessor,
- SecurityService securityService) throws IOException {
+ CacheServerStats stats, int hsTimeout, int socketBufferSize, String communicationModeStr,
+ byte communicationMode, Acceptor acceptor, ClientProtocolProcessor clientProtocolProcessor,
+ SecurityService securityService) throws IOException {
super(socket, c, helper, stats, hsTimeout, socketBufferSize, communicationModeStr,
communicationMode, acceptor, securityService);
this.protocolProcessor = clientProtocolProcessor;
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
index 0707e73..2f090dc 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnection.java
@@ -73,6 +73,7 @@ import org.apache.geode.internal.util.Breadcrumbs;
import org.apache.geode.security.AuthenticationFailedException;
import org.apache.geode.security.AuthenticationRequiredException;
import org.apache.geode.security.GemFireSecurityException;
+import org.apache.geode.statistics.cache.CacheServerStats;
/**
* Provides an implementation for the server socket end of the hierarchical cache connection. Each
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java
index 2c519a3..f30ab26 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ServerConnectionFactory.java
@@ -20,7 +20,6 @@ import static org.apache.geode.internal.cache.tier.CommunicationMode.ProtobufCli
import java.io.IOException;
import java.net.Socket;
-import org.apache.geode.statistics.StatisticsFactory;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.client.protocol.ClientProtocolProcessor;
import org.apache.geode.internal.cache.client.protocol.ClientProtocolService;
@@ -30,6 +29,7 @@ import org.apache.geode.internal.cache.client.protocol.exception.ServiceVersionN
import org.apache.geode.internal.cache.tier.Acceptor;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.security.SecurityService;
+import org.apache.geode.statistics.cache.CacheServerStats;
/**
* Creates instances of ServerConnection based on the connection mode provided.
@@ -43,19 +43,18 @@ public class ServerConnectionFactory {
}
- private synchronized ClientProtocolService getClientProtocolService(
- StatisticsFactory statisticsFactory, String serverName) {
+ private synchronized ClientProtocolService getClientProtocolService(String serverName) {
if (clientProtocolService == null) {
clientProtocolService = clientProtocolServiceLoader.lookupService();
- clientProtocolService.initializeStatistics(serverName, statisticsFactory);
+ clientProtocolService.initializeStatistics(serverName);
}
return clientProtocolService;
}
public ServerConnection makeServerConnection(Socket socket, InternalCache cache,
- CachedRegionHelper helper, CacheServerStats stats, int hsTimeout, int socketBufferSize,
- String communicationModeStr, byte communicationMode, Acceptor acceptor,
- SecurityService securityService) throws IOException {
+ CachedRegionHelper helper, CacheServerStats stats, int hsTimeout, int socketBufferSize,
+ String communicationModeStr, byte communicationMode, Acceptor acceptor,
+ SecurityService securityService) throws IOException {
if (ProtobufClientServerProtocol.getModeNumber() == communicationMode) {
if (!Boolean.getBoolean("geode.feature-protobuf-protocol")) {
throw new IOException("Server received unknown communication mode: " + communicationMode);
@@ -75,12 +74,11 @@ public class ServerConnectionFactory {
}
}
- private ServerConnection createProtobufServerConnection(Socket socket, InternalCache cache,
- CachedRegionHelper helper, CacheServerStats stats, int hsTimeout, int socketBufferSize,
- String communicationModeStr, byte communicationMode, Acceptor acceptor,
- SecurityService securityService) throws IOException {
- ClientProtocolService service =
- getClientProtocolService(cache.getDistributedSystem().getStatisticsFactory(), acceptor.getServerName());
+ private ProtobufServerConnection createProtobufServerConnection(Socket socket, InternalCache cache,
+ CachedRegionHelper helper, CacheServerStats stats, int hsTimeout, int socketBufferSize,
+ String communicationModeStr, byte communicationMode, Acceptor acceptor,
+ SecurityService securityService) throws IOException {
+ ClientProtocolService service = getClientProtocolService(acceptor.getServerName());
ClientProtocolProcessor processor = service.createProcessorForCache(cache, securityService);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
index 258ac30..e1ea2ba 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
@@ -53,7 +53,6 @@ import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ResourceEvent;
import org.apache.geode.distributed.internal.ServerLocation;
-import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.EnumListenerEvent;
import org.apache.geode.internal.cache.HasCachePerfStats;
@@ -77,6 +76,8 @@ import org.apache.geode.internal.offheap.Releasable;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.offheap.annotations.Unretained;
+import org.apache.geode.statistics.cache.CachePerfStats;
+import org.apache.geode.statistics.wan.GatewaySenderStats;
/**
* Abstract implementation of both Serial and Parallel GatewaySender. It handles common
@@ -266,7 +267,7 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
this.stopper = new Stopper(cache.getCancelCriterion());
this.senderAdvisor = GatewaySenderAdvisor.createGatewaySenderAdvisor(this);
if (!this.isForInternalUse()) {
- this.statistics = new GatewaySenderStats(cache.getDistributedSystem().getStatisticsFactory(), id);
+ this.statistics = new GatewaySenderStats(id);
}
initializeEventIdIndex();
}
@@ -1205,7 +1206,7 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
// Create a stats holder for the meta data stats
final HasCachePerfStats statsHolder = new HasCachePerfStats() {
public CachePerfStats getCachePerfStats() {
- return new CachePerfStats(cache.getDistributedSystem().getStatisticsFactory(), META_DATA_REGION_NAME);
+ return new CachePerfStats(META_DATA_REGION_NAME);
}
};
diff --git a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/cache/CacheServerStats.kt b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/cache/CacheServerStats.kt
index 2ca6ef7..fe20d90 100644
--- a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/cache/CacheServerStats.kt
+++ b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/cache/CacheServerStats.kt
@@ -21,129 +21,131 @@ import org.apache.geode.statistics.internal.micrometer.GaugeStatisticMeter
import org.apache.geode.statistics.internal.micrometer.MicrometerMeterGroup
import org.apache.geode.statistics.internal.micrometer.TimerStatisticMeter
-class CacheServerStats(ownerName: String, typeName: String = "CacheServerStats") : MicrometerMeterGroup("$typeName-$ownerName"), MessageStats {
-
- private val cacheServerClientGetRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client get requests.", arrayOf("owner", ownerName, "operation", "get"))
- private val cacheServerClientGetRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading get requests.", arrayOf("owner", ownerName, "operation", "get"), unit = "nanoseconds")
- private val cacheServerGetProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client get request, including the time to get an object from the cache.", arrayOf("owner", ownerName, "operation", "get"), unit = "nanoseconds")
- private val cacheServerGetResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of get responses written to the cache client.", arrayOf("owner", ownerName, "operation", "get"))
- private val cacheServerGetResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing get responses.", arrayOf("owner", ownerName, "operation", "get"), unit = "nanoseconds")
-
- private val cacheServerClientPutRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client put requests.", arrayOf("owner", ownerName, "operation", "put"))
- private val cacheServerClientPutRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading put requests.", arrayOf("owner", ownerName, "operation", "put"), unit = "nanoseconds")
- private val cacheServerPutProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client put request, including the time to put an object into the cache.", arrayOf("owner", ownerName, "operation", "put"), unit = "nanoseconds")
- private val cacheServerPutResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of put responses written to the cache client.", arrayOf("owner", ownerName, "operation", "put"))
- private val cacheServerPutResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing put responses.", arrayOf("owner", ownerName, "operation", "put"), unit = "nanoseconds")
-
- private val cacheServerClientPutAllRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client putAll requests.", arrayOf("owner", ownerName, "operation", "putAll"))
- private val cacheServerClientPutAllRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading putAll requests.", arrayOf("owner", ownerName, "operation", "putAll"), unit = "nanoseconds")
- private val cacheServerPutAllProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client putAll request, including the time to put all objects into the cache.", arrayOf("owner", ownerName, "operation", "putAll"), unit = "nanoseconds")
- private val cacheServerPutAllResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of putAll responses written to the cache client.", arrayOf("owner", ownerName, "operation", "putAll"))
- private val cacheServerPutAllResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing putAll responses.", arrayOf("owner", ownerName, "operation", "putAll"), unit = "nanoseconds")
-
- private val cacheServerClientRemoveAllRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client removeAll requests.", arrayOf("owner", ownerName, "operation", "removeAll"))
- private val cacheServerClientRemoveAllRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading removeAll requests.", arrayOf("owner", ownerName, "operation", "removeAll"), unit = "nanoseconds")
- private val cacheServerRemoveAllProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client removeAll request, including the time to remove all objects from the cache.", arrayOf("owner", ownerName, "operation", "removeAll"), unit = "nanoseconds")
- private val cacheServerRemoveAllResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of removeAll responses written to the cache client.", arrayOf("owner", ownerName, "operation", "removeAll"))
- private val cacheServerRemoveAllResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing removeAll responses.", arrayOf("owner", ownerName, "operation", "removeAll"), unit = "nanoseconds")
-
- private val cacheServerClientGetAllRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client getAll requests.", arrayOf("owner", ownerName, "operation", "getAll"))
- private val cacheServerClientGetAllRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading getAll requests.", arrayOf("owner", ownerName, "operation", "getAll"), unit = "nanoseconds")
- private val cacheServerGetAllProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client getAll request.", arrayOf("owner", ownerName, "operation", "getAll"), unit = "nanoseconds")
- private val cacheServerGetAllResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of getAll responses written to the cache client.", arrayOf("owner", ownerName, "operation", "getAll"))
- private val cacheServerGetAllResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing getAll responses.", arrayOf("owner", ownerName, "operation", "getAll"), unit = "nanoseconds")
-
- private val cacheServerClientDestroyRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client destroy requests.", arrayOf("owner", ownerName, "operation", "destroy"))
- private val cacheServerClientDestroyRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading destroy requests.", arrayOf("owner", ownerName, "operation", "destroy"), unit = "nanoseconds")
- private val cacheServerDestroyProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client destroy request, including the time to destroy an object from the cache.", arrayOf("owner", ownerName, "operation", "destroy"), unit = "nanoseconds")
- private val cacheServerDestroyResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of destroy responses written to the cache client.", arrayOf("owner", ownerName, "operation", "destroy"))
- private val cacheServerDestroyResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing destroy responses.", arrayOf("owner", ownerName, "operation", "destroy"), unit = "nanoseconds")
-
- private val cacheServerClientInvaldiateRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client invalidate requests.", arrayOf("owner", ownerName, "operation", "invalidate"))
- private val cacheServerClientInvaldiateRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading invalidate requests.", arrayOf("owner", ownerName, "operation", "invalidate"), unit = "nanoseconds")
- private val cacheServerInvaldiateProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client invalidate request, including the time to invalidate an object from the cache.", arrayOf("owner", ownerName, "operation", "invalidate"), unit = "nanoseconds")
- private val cacheServerInvaldiateResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of invalidate responses written to the cache client.", arrayOf("owner", ownerName, "operation", "invalidate"))
- private val cacheServerInvaldiateResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing invalidate responses.", arrayOf("owner", ownerName, "operation", "invalidate"), unit = "nanoseconds")
-
- private val cacheServerClientSizeRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client size requests.", arrayOf("owner", ownerName, "operation", "size"))
- private val cacheServerClientSizeRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading size requests.", arrayOf("owner", ownerName, "operation", "size"), unit = "nanoseconds")
- private val cacheServerSizeProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client size request, including the time to size an object from the cache.", arrayOf("owner", ownerName, "operation", "size"), unit = "nanoseconds")
- private val cacheServerSizeResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of size responses written to the cache client.", arrayOf("owner", ownerName, "operation", "size"))
- private val cacheServerSizeResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing size responses.", arrayOf("owner", ownerName, "operation", "size"), unit = "nanoseconds")
-
- private val cacheServerClientQueryRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client query requests.", arrayOf("owner", ownerName, "operation", "query"))
- private val cacheServerClientQueryRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading query requests.", arrayOf("owner", ownerName, "operation", "query"), unit = "nanoseconds")
- private val cacheServerQueryProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client query request, including the time to destroy an object from the cache.", arrayOf("owner", ownerName, "operation", "query"), unit = "nanoseconds")
- private val cacheServerQueryResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of query responses written to the cache client.", arrayOf("owner", ownerName, "operation", "query"))
- private val cacheServerQueryResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing query responses.", arrayOf("owner", ownerName, "operation", "query"), unit = "nanoseconds")
-
- private val cacheServerClientDestroyRegionRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client destroyRegion requests.", arrayOf("owner", ownerName, "operation", "destroyRegion"))
- private val cacheServerClientDestroyRegionRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading destroyRegion requests.", arrayOf("owner", ownerName, "operation", "destroyRegion"), unit = "nanoseconds")
- private val cacheServerClientDestroyRegionProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client destroyRegion request, including the time to destroy the region from the cache.", arrayOf("owner", ownerName, "operation", "destroyRegion"), unit = "nanoseconds")
- private val cacheServerClientDestroyRegionResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of destroyRegion responses written to the cache client.", arrayOf("owner", ownerName, "operation", "destroyRegion"))
- private val cacheServerClientDestroyRegionResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing destroyRegion responses.", arrayOf("owner", ownerName, "operation", "destroyRegion"), unit = "nanoseconds")
-
- private val cacheServerClientContainsKeyRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client containsKey requests.", arrayOf("owner", ownerName, "operation", "containsKey"))
- private val cacheServerClientContainsKeyRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent reading containsKey requests.", arrayOf("owner", ownerName, "operation", "containsKey"), unit = "nanoseconds")
- private val cacheServerClientContainsKeyProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent processing a containsKey request.", arrayOf("owner", ownerName, "operation", "containsKey"), unit = "nanoseconds")
- private val cacheServerClientContainsKeyResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of containsKey responses written to the cache client.", arrayOf("owner", ownerName, "operation", "containsKey"))
- private val cacheServerClientContainsKeyResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent writing containsKey responses.", arrayOf("owner", ownerName, "operation", "containsKey"), unit = "nanoseconds")
-
- private val cacheServerProcessBatchRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client processBatch requests.", arrayOf("owner", ownerName, "operation", "processBatch"))
- private val cacheServerProcessBatchRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading processBatch requests.", arrayOf("owner", ownerName, "operation", "processBatch"), unit = "nanoseconds")
- private val cacheServerProcessBatchProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client processBatch request.", arrayOf("owner", ownerName, "operation", "processBatch"), unit = "nanoseconds")
- private val cacheServerProcessBatchResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of processBatch responses written to the cache client.", arrayOf("owner", ownerName, "operation", "processBatch"))
- private val cacheServerProcessBatchResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing processBatch responses.", arrayOf("owner", ownerName, "operation", "processBatch"), unit = "nanoseconds")
- private val cacheServerProcessBatchSizeMeter = CounterStatisticMeter("cacheserver.client.batch.size", "The size of the batches received.", arrayOf("owner", ownerName, "operation", "processBatch"), unit = "bytes")
-
- private val cacheServerClientClearRegionRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client clearRegion requests.", arrayOf("owner", ownerName, "operation", "clearRegion"))
- private val cacheServerClientClearRegionRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading clearRegion requests.", arrayOf("owner", ownerName, "operation", "clearRegion"), unit = "nanoseconds")
- private val cacheServerClearRegionProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client clearRegion request, including the time to destroy the region from the cache.", arrayOf("owner", ownerName, "operation", "clearRegion"), unit = "nanoseconds")
- private val cacheServerClearRegionResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of clearRegion responses written to the cache client.", arrayOf("owner", ownerName, "operation", "clearRegion"))
- private val cacheServerClearRegionResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing clearRegion responses.", arrayOf("owner", ownerName, "operation", "clearRegion"), unit = "nanoseconds")
-
-
- private val cacheServerClientNotificationRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client notification requests.", arrayOf("owner", ownerName, "operation", "clientNotification"))
- private val cacheServerClientNotificationRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading client notification requests.", arrayOf("owner", ownerName, "operation", "clientNotification"), unit = "nanoseconds")
- private val cacheServerClientNotificationProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client notification request.", arrayOf("owner", ownerName, "operation", "clientNotification"), unit = "nanoseconds")
-
- private val cacheServerUpdateClientNotificationRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client notification update requests.", arrayOf("owner", ownerName, "operation", "updateClientNotification"))
- private val cacheServerUpdateClientNotificationRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading client notification update requests.", arrayOf("owner", ownerName, "operation", "updateClientNotification"), unit = "nanoseconds")
- private val cacheServerUpdateClientNotificationProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a client notification update request.", arrayOf("owner", ownerName, "operation", "updateClientNotification"), unit = "nanoseconds")
-
- private val cacheServerClientReadyRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client ready requests.", arrayOf("owner", ownerName, "operation", "clientReady"))
- private val cacheServerClientReadyRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading cache client ready requests.", arrayOf("owner", ownerName, "operation", "clientReady"), unit = "nanoseconds")
- private val cacheServerClientReadyProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client ready request, including the time to destroy an object from the cache.", arrayOf("owner", ownerName, "operation", "clientReady"), unit = "nanoseconds")
- private val cacheServerClientReadyResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of client ready responses written to the cache client.", arrayOf("owner", ownerName, "operation", "clientReady"))
- private val cacheServerClientReadyResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing client ready responses.", arrayOf("owner", ownerName, "operation", "clientReady"), unit = "nanoseconds")
-
- private val cacheServerClientCloseConnectionRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client close connection requests.", arrayOf("owner", ownerName, "operation", "closeConnection"))
- private val cacheServerClientCloseConnectionRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading close connection requests.", arrayOf("owner", ownerName, "operation", "closeConnection"), unit = "nanoseconds")
- private val cacheServerClientCloseConnectionProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client close connection request.", arrayOf("owner", ownerName, "operation", "closeConnection"), unit = "nanoseconds")
-
-
- private val cacheServerFailedConnectionAttemptsMeter = CounterStatisticMeter("cacheserver.client.connection.failed.count", "Number of failed connection attempts.", arrayOf("owner", ownerName))
- private val cacheServerCurrentConnectionMeter = GaugeStatisticMeter("cacheserver.client.connection.count", "Number of sockets accepted and used for client to server messaging.", arrayOf("owner", ownerName, "connectionType", "messaging"))
- private val cacheServerCurrentQueueConnectionMeter = GaugeStatisticMeter("cacheserver.client.connection.count", "Number of sockets accepted and used for server to client queue messaging.", arrayOf("owner", ownerName, "connectionType", "queue"))
- private val cacheServerClientCountMeter = GaugeStatisticMeter("cacheserver.client.count", "Number of client virtual machines connected.", arrayOf("owner", ownerName))
- private val cacheServerOutOfOrderGatewayBatchIdMeter = CounterStatisticMeter("cacheserver.gateway.batch.outoforder.count", "Number of Out of order batch IDs.", arrayOf("owner", ownerName))
-
- private val cacheServerClientWriteRequestAbandondMeter = CounterStatisticMeter("cacheserver.client.requests.abandond.count", "Number of write opertations abandond by clients", arrayOf("owner", ownerName, "requestType", "write"))
- private val cacheServerClientReadRequestAbandondMeter = CounterStatisticMeter("cacheserver.client.requests.abandond.count", "Number of read opertations abandond by clients", arrayOf("owner", ownerName, "requestType", "read"))
- private val cacheServerReceivedBytesMeter = CounterStatisticMeter("cacheserver.client.bytes", "Total number of bytes received from clients.", arrayOf("owner", ownerName, "direction", "received"), unit = "bytes")
- private val cacheServerSentBytesMeter = CounterStatisticMeter("cacheserver.client.bytes", "Total number of bytes sent to clients.", arrayOf("owner", ownerName, "direction", "sent"), unit = "bytes")
- private val cacheServerMessagesReceivedMeter = GaugeStatisticMeter("cacheserver.client.messages.received.count", "Current number of message being received off the network or being processed after reception.", arrayOf("owner", ownerName))
- private val cacheServerMessagesReceivedBytesMeter = GaugeStatisticMeter("cacheserver.client.messages.received.bytes", "Current number of bytes consumed by messages being received or processed.", arrayOf("owner", ownerName), unit = "bytes")
- private val cacheServerConnectionTimeoutMeter = CounterStatisticMeter("cacheserver.client.connection.timeout.count", "Total number of connections that have been timed out by the server because of client inactivity", arrayOf("owner", ownerName))
- private val cacheServerThreadQueueSizeMeter = GaugeStatisticMeter("cacheserver.client.thread.queue.size", "Current number of connections waiting for a thread to start processing their message.", arrayOf("owner", ownerName))
- private val cacheServerConnectionAcceptInProgressMeter = GaugeStatisticMeter("cacheserver.client.connection.accept.inprogress.count", "Current number of server accepts that are attempting to do the initial handshake with the client.", arrayOf("owner", ownerName))
- private val cacheServerConnectionAcceptStartMeter = CounterStatisticMeter("cacheserver.client.connection.accept.start.count", "Total number of threads created to deal with an accepted socket. Note that this is not the current number of threads.", arrayOf("owner", ownerName))
- private val cacheServerConnectionStartMeter = CounterStatisticMeter("cacheserver.client.connection.start.count", "Total number of threads created to deal with a client connection. Note that this is not the current number of threads.", arrayOf("owner", ownerName))
- private val cacheServerConnectionThreadsMeter = GaugeStatisticMeter("cacheserver.client.connection.thread.count", "Current number of threads dealing with a client connection.", arrayOf("owner", ownerName))
- private val cacheServerConnectionLoadMeter = GaugeStatisticMeter("cacheserver.client.connection.load.count", "The load from client to server connections as reported by the load probe installed in this server", arrayOf("owner", ownerName))
- private val cacheServerLoadPerConnectionMeter = GaugeStatisticMeter("cacheserver.client.load.connection.count", "The estimate of how much load is added for each new connection as reported by the load probe installed in this server", arrayOf("owner", ownerName))
- private val cacheServerQueueLoadMeter = GaugeStatisticMeter("cacheserver.client.queue.load.count", "The load from queues as reported by the load probe installed in this server", arrayOf("owner", ownerName))
- private val cacheServerLoadPerQueueMeter = GaugeStatisticMeter("cacheserver.client.load.queue.count", "The estimate of how much load is added for each new connection as reported by the load probe installed in this server", arrayOf("owner", ownerName))
+open class CacheServerStats @JvmOverloads constructor(private val ownerName: String, typeName: String = "CacheServerStats-$ownerName") : MicrometerMeterGroup("$typeName-$ownerName"), MessageStats {
+
+ override fun getCommonTags(): Array<String> = arrayOf("owner", ownerName)
+
+ private val cacheServerClientGetRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client get requests.", arrayOf( "operation", "get"))
+ private val cacheServerClientGetRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading get requests.", arrayOf( "operation", "get"), unit = "nanoseconds")
+ private val cacheServerGetProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client get request, including the time to get an object from the cache.", arrayOf( "operation", "get"), unit = "nanoseconds")
+ private val cacheServerGetResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of get responses written to the cache client.", arrayOf( "operation", "get"))
+ private val cacheServerGetResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing get responses.", arrayOf( "operation", "get"), unit = "nanoseconds")
+
+ private val cacheServerClientPutRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client put requests.", arrayOf( "operation", "put"))
+ private val cacheServerClientPutRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading put requests.", arrayOf( "operation", "put"), unit = "nanoseconds")
+ private val cacheServerPutProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client put request, including the time to put an object into the cache.", arrayOf( "operation", "put"), unit = "nanoseconds")
+ private val cacheServerPutResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of put responses written to the cache client.", arrayOf( "operation", "put"))
+ private val cacheServerPutResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing put responses.", arrayOf( "operation", "put"), unit = "nanoseconds")
+
+ private val cacheServerClientPutAllRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client putAll requests.", arrayOf( "operation", "putAll"))
+ private val cacheServerClientPutAllRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading putAll requests.", arrayOf( "operation", "putAll"), unit = "nanoseconds")
+ private val cacheServerPutAllProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client putAll request, including the time to put all objects into the cache.", arrayOf( "operation", "putAll"), unit = "nanoseconds")
+ private val cacheServerPutAllResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of putAll responses written to the cache client.", arrayOf( "operation", "putAll"))
+ private val cacheServerPutAllResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing putAll responses.", arrayOf( "operation", "putAll"), unit = "nanoseconds")
+
+ private val cacheServerClientRemoveAllRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client removeAll requests.", arrayOf( "operation", "removeAll"))
+ private val cacheServerClientRemoveAllRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading removeAll requests.", arrayOf( "operation", "removeAll"), unit = "nanoseconds")
+ private val cacheServerRemoveAllProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client removeAll request, including the time to remove all objects from the cache.", arrayOf( "operation", "removeAll"), unit = "nanoseconds")
+ private val cacheServerRemoveAllResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of removeAll responses written to the cache client.", arrayOf( "operation", "removeAll"))
+ private val cacheServerRemoveAllResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing removeAll responses.", arrayOf( "operation", "removeAll"), unit = "nanoseconds")
+
+ private val cacheServerClientGetAllRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client getAll requests.", arrayOf( "operation", "getAll"))
+ private val cacheServerClientGetAllRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading getAll requests.", arrayOf( "operation", "getAll"), unit = "nanoseconds")
+ private val cacheServerGetAllProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client getAll request.", arrayOf( "operation", "getAll"), unit = "nanoseconds")
+ private val cacheServerGetAllResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of getAll responses written to the cache client.", arrayOf( "operation", "getAll"))
+ private val cacheServerGetAllResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing getAll responses.", arrayOf( "operation", "getAll"), unit = "nanoseconds")
+
+ private val cacheServerClientDestroyRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client destroy requests.", arrayOf( "operation", "destroy"))
+ private val cacheServerClientDestroyRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading destroy requests.", arrayOf( "operation", "destroy"), unit = "nanoseconds")
+ private val cacheServerDestroyProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client destroy request, including the time to destroy an object from the cache.", arrayOf( "operation", "destroy"), unit = "nanoseconds")
+ private val cacheServerDestroyResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of destroy responses written to the cache client.", arrayOf( "operation", "destroy"))
+ private val cacheServerDestroyResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing destroy responses.", arrayOf( "operation", "destroy"), unit = "nanoseconds")
+
+ private val cacheServerClientInvaldiateRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client invalidate requests.", arrayOf( "operation", "invalidate"))
+ private val cacheServerClientInvaldiateRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading invalidate requests.", arrayOf( "operation", "invalidate"), unit = "nanoseconds")
+ private val cacheServerInvaldiateProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client invalidate request, including the time to invalidate an object from the cache.", arrayOf( "operation", "invalidate"), unit = "nanoseconds")
+ private val cacheServerInvaldiateResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of invalidate responses written to the cache client.", arrayOf( "operation", "invalidate"))
+ private val cacheServerInvaldiateResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing invalidate responses.", arrayOf( "operation", "invalidate"), unit = "nanoseconds")
+
+ private val cacheServerClientSizeRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client size requests.", arrayOf( "operation", "size"))
+ private val cacheServerClientSizeRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading size requests.", arrayOf( "operation", "size"), unit = "nanoseconds")
+ private val cacheServerSizeProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client size request, including the time to size an object from the cache.", arrayOf( "operation", "size"), unit = "nanoseconds")
+ private val cacheServerSizeResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of size responses written to the cache client.", arrayOf( "operation", "size"))
+ private val cacheServerSizeResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing size responses.", arrayOf( "operation", "size"), unit = "nanoseconds")
+
+ private val cacheServerClientQueryRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client query requests.", arrayOf( "operation", "query"))
+ private val cacheServerClientQueryRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading query requests.", arrayOf( "operation", "query"), unit = "nanoseconds")
+ private val cacheServerQueryProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client query request, including the time to destroy an object from the cache.", arrayOf( "operation", "query"), unit = "nanoseconds")
+ private val cacheServerQueryResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of query responses written to the cache client.", arrayOf( "operation", "query"))
+ private val cacheServerQueryResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing query responses.", arrayOf( "operation", "query"), unit = "nanoseconds")
+
+ private val cacheServerClientDestroyRegionRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client destroyRegion requests.", arrayOf( "operation", "destroyRegion"))
+ private val cacheServerClientDestroyRegionRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading destroyRegion requests.", arrayOf( "operation", "destroyRegion"), unit = "nanoseconds")
+ private val cacheServerClientDestroyRegionProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client destroyRegion request, including the time to destroy the region from the cache.", arrayOf( "operation", "destroyRegion"), unit = "nanoseconds")
+ private val cacheServerClientDestroyRegionResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of destroyRegion responses written to the cache client.", arrayOf( "operation", "destroyRegion"))
+ private val cacheServerClientDestroyRegionResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing destroyRegion responses.", arrayOf( "operation", "destroyRegion"), unit = "nanoseconds")
+
+ private val cacheServerClientContainsKeyRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client containsKey requests.", arrayOf( "operation", "containsKey"))
+ private val cacheServerClientContainsKeyRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent reading containsKey requests.", arrayOf( "operation", "containsKey"), unit = "nanoseconds")
+ private val cacheServerClientContainsKeyProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent processing a containsKey request.", arrayOf( "operation", "containsKey"), unit = "nanoseconds")
+ private val cacheServerClientContainsKeyResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of containsKey responses written to the cache client.", arrayOf( "operation", "containsKey"))
+ private val cacheServerClientContainsKeyResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent writing containsKey responses.", arrayOf( "operation", "containsKey"), unit = "nanoseconds")
+
+ private val cacheServerProcessBatchRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client processBatch requests.", arrayOf( "operation", "processBatch"))
+ private val cacheServerProcessBatchRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading processBatch requests.", arrayOf( "operation", "processBatch"), unit = "nanoseconds")
+ private val cacheServerProcessBatchProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client processBatch request.", arrayOf( "operation", "processBatch"), unit = "nanoseconds")
+ private val cacheServerProcessBatchResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of processBatch responses written to the cache client.", arrayOf( "operation", "processBatch"))
+ private val cacheServerProcessBatchResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing processBatch responses.", arrayOf( "operation", "processBatch"), unit = "nanoseconds")
+ private val cacheServerProcessBatchSizeMeter = CounterStatisticMeter("cacheserver.client.batch.size", "The size of the batches received.", arrayOf( "operation", "processBatch"), unit = "bytes")
+
+ private val cacheServerClientClearRegionRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client clearRegion requests.", arrayOf( "operation", "clearRegion"))
+ private val cacheServerClientClearRegionRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading clearRegion requests.", arrayOf( "operation", "clearRegion"), unit = "nanoseconds")
+ private val cacheServerClearRegionProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client clearRegion request, including the time to destroy the region from the cache.", arrayOf( "operation", "clearRegion"), unit = "nanoseconds")
+ private val cacheServerClearRegionResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of clearRegion responses written to the cache client.", arrayOf( "operation", "clearRegion"))
+ private val cacheServerClearRegionResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing clearRegion responses.", arrayOf( "operation", "clearRegion"), unit = "nanoseconds")
+
+
+ private val cacheServerClientNotificationRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client notification requests.", arrayOf( "operation", "clientNotification"))
+ private val cacheServerClientNotificationRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading client notification requests.", arrayOf( "operation", "clientNotification"), unit = "nanoseconds")
+ private val cacheServerClientNotificationProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client notification request.", arrayOf( "operation", "clientNotification"), unit = "nanoseconds")
+
+ private val cacheServerUpdateClientNotificationRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client notification update requests.", arrayOf( "operation", "updateClientNotification"))
+ private val cacheServerUpdateClientNotificationRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading client notification update requests.", arrayOf( "operation", "updateClientNotification"), unit = "nanoseconds")
+ private val cacheServerUpdateClientNotificationProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a client notification update request.", arrayOf( "operation", "updateClientNotification"), unit = "nanoseconds")
+
+ private val cacheServerClientReadyRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client ready requests.", arrayOf( "operation", "clientReady"))
+ private val cacheServerClientReadyRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading cache client ready requests.", arrayOf( "operation", "clientReady"), unit = "nanoseconds")
+ private val cacheServerClientReadyProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client ready request, including the time to destroy an object from the cache.", arrayOf( "operation", "clientReady"), unit = "nanoseconds")
+ private val cacheServerClientReadyResponseWrittenMeter = CounterStatisticMeter("cacheserver.client.responses.written.count", "Number of client ready responses written to the cache client.", arrayOf( "operation", "clientReady"))
+ private val cacheServerClientReadyResponseWrittenTimer = TimerStatisticMeter("cacheserver.client.responses.written.time", "Total time spent in writing client ready responses.", arrayOf( "operation", "clientReady"), unit = "nanoseconds")
+
+ private val cacheServerClientCloseConnectionRequestMeter = CounterStatisticMeter("cacheserver.client.requests.count", "Number of cache client close connection requests.", arrayOf( "operation", "closeConnection"))
+ private val cacheServerClientCloseConnectionRequestTimer = TimerStatisticMeter("cacheserver.client.requests.time", "Total time spent in reading close connection requests.", arrayOf( "operation", "closeConnection"), unit = "nanoseconds")
+ private val cacheServerClientCloseConnectionProcessTimer = TimerStatisticMeter("cacheserver.operation.process.time", "Total time spent in processing a cache client close connection request.", arrayOf( "operation", "closeConnection"), unit = "nanoseconds")
+
+
+ private val cacheServerFailedConnectionAttemptsMeter = CounterStatisticMeter("cacheserver.client.connection.failed.count", "Number of failed connection attempts.")
+ private val cacheServerCurrentConnectionMeter = GaugeStatisticMeter("cacheserver.client.connection.count", "Number of sockets accepted and used for client to server messaging.", arrayOf( "connectionType", "messaging"))
+ private val cacheServerCurrentQueueConnectionMeter = GaugeStatisticMeter("cacheserver.client.connection.count", "Number of sockets accepted and used for server to client queue messaging.", arrayOf( "connectionType", "queue"))
+ private val cacheServerClientCountMeter = GaugeStatisticMeter("cacheserver.client.count", "Number of client virtual machines connected.")
+ private val cacheServerOutOfOrderGatewayBatchIdMeter = CounterStatisticMeter("cacheserver.gateway.batch.outoforder.count", "Number of Out of order batch IDs.")
+
+ private val cacheServerClientWriteRequestAbandondMeter = CounterStatisticMeter("cacheserver.client.requests.abandond.count", "Number of write opertations abandond by clients", arrayOf( "requestType", "write"))
+ private val cacheServerClientReadRequestAbandondMeter = CounterStatisticMeter("cacheserver.client.requests.abandond.count", "Number of read opertations abandond by clients", arrayOf( "requestType", "read"))
+ private val cacheServerReceivedBytesMeter = CounterStatisticMeter("cacheserver.client.bytes", "Total number of bytes received from clients.", arrayOf( "direction", "received"), unit = "bytes")
+ private val cacheServerSentBytesMeter = CounterStatisticMeter("cacheserver.client.bytes", "Total number of bytes sent to clients.", arrayOf( "direction", "sent"), unit = "bytes")
+ private val cacheServerMessagesReceivedMeter = GaugeStatisticMeter("cacheserver.client.messages.received.count", "Current number of message being received off the network or being processed after reception.")
+ private val cacheServerMessagesReceivedBytesMeter = GaugeStatisticMeter("cacheserver.client.messages.received.bytes", "Current number of bytes consumed by messages being received or processed.", unit = "bytes")
+ private val cacheServerConnectionTimeoutMeter = CounterStatisticMeter("cacheserver.client.connection.timeout.count", "Total number of connections that have been timed out by the server because of client inactivity")
+ private val cacheServerThreadQueueSizeMeter = GaugeStatisticMeter("cacheserver.client.thread.queue.size", "Current number of connections waiting for a thread to start processing their message.")
+ private val cacheServerConnectionAcceptInProgressMeter = GaugeStatisticMeter("cacheserver.client.connection.accept.inprogress.count", "Current number of server accepts that are attempting to do the initial handshake with the client.")
+ private val cacheServerConnectionAcceptStartMeter = CounterStatisticMeter("cacheserver.client.connection.accept.start.count", "Total number of threads created to deal with an accepted socket. Note that this is not the current number of threads.")
+ private val cacheServerConnectionStartMeter = CounterStatisticMeter("cacheserver.client.connection.start.count", "Total number of threads created to deal with a client connection. Note that this is not the current number of threads.")
+ private val cacheServerConnectionThreadsMeter = GaugeStatisticMeter("cacheserver.client.connection.thread.count", "Current number of threads dealing with a client connection.")
+ private val cacheServerConnectionLoadMeter = GaugeStatisticMeter("cacheserver.client.connection.load.count", "The load from client to server connections as reported by the load probe installed in this server")
+ private val cacheServerLoadPerConnectionMeter = GaugeStatisticMeter("cacheserver.client.load.connection.count", "The estimate of how much load is added for each new connection as reported by the load probe installed in this server")
+ private val cacheServerQueueLoadMeter = GaugeStatisticMeter("cacheserver.client.queue.load.count", "The load from queues as reported by the load probe installed in this server")
+ private val cacheServerLoadPerQueueMeter = GaugeStatisticMeter("cacheserver.client.load.queue.count", "The estimate of how much load is added for each new connection as reported by the load probe installed in this server")
override fun initializeStaticMeters() {
diff --git a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/distributed/DistributionStats.kt b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/distributed/DistributionStats.kt
index e7134cf..ceb95ba 100644
--- a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/distributed/DistributionStats.kt
+++ b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/distributed/DistributionStats.kt
@@ -10,7 +10,9 @@ import org.apache.geode.statistics.internal.micrometer.MicrometerMeterGroup
import org.apache.geode.statistics.internal.micrometer.TimerStatisticMeter
import org.apache.geode.statistics.util.NOW_NANOS
-class DistributionStats : MicrometerMeterGroup("DistributionStats"), DMStats {
+class DistributionStats(private val processID: Long) : MicrometerMeterGroup("DistributionStats-$processID"), DMStats {
+ override fun getCommonTags(): Array<String> = arrayOf("processID", processID.toString())
+
override fun initializeStaticMeters() {
registerMeter(distributionMessageMeter)
registerMeter(distributionMessageTransactionCommitMeter)
@@ -1000,6 +1002,9 @@ class DistributionStats : MicrometerMeterGroup("DistributionStats"), DMStats {
udpFinalCheckResponseReceivedMeter.increment()
}
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ fun getSerialQueueBytes(): Long = distributionQueueSerialBytesMeter.getValue()
+
val overflowQueueHelper: ThrottledQueueStatHelper
get() = object : ThrottledQueueStatHelper {
override fun incThrottleCount() {
diff --git a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/distributed/ThrottledMemQueueStatHelper.kt b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/distributed/ThrottledMemQueueStatHelper.kt
deleted file mode 100644
index 30f2b78..0000000
--- a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/distributed/ThrottledMemQueueStatHelper.kt
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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
-
-/**
- * Used to implement statistics on a throttled queue. The implementation will call these methods at
- * to proper time.
- *
- *
- * @since GemFire 5.0
- */
-interface ThrottledMemQueueStatHelper : QueueStatHelper {
-
- /**
- * Called each time a thread was delayed by the throttle.
- */
- fun incThrottleCount()
-
- /**
- * Called after a throttled operation has completed.
- *
- * @param nanos the amount of time, in nanoseconds, the throttle caused us to wait.
- */
- fun throttleTime(nanos: Long)
-
- /**
- * Increments the amount of memory consumed by queue contents.
- *
- * @param amount number of bytes added to the queue
- */
- fun addMem(amount: Int)
-
- /**
- * Decrements the amount of memory consumed by queue contents.
- *
- * @param amount number of bytes removed from the queue
- */
- fun removeMem(amount: Int)
-}
diff --git a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/dlock/DLockStats.kt b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/dlock/DLockStats.kt
new file mode 100755
index 0000000..e1ceb8e
--- /dev/null
+++ b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/dlock/DLockStats.kt
@@ -0,0 +1,651 @@
+/*
+ * 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.statistics.dlock
+
+import org.apache.geode.distributed.internal.PoolStatHelper
+import org.apache.geode.distributed.internal.QueueStatHelper
+import org.apache.geode.statistics.internal.micrometer.CounterStatisticMeter
+import org.apache.geode.statistics.internal.micrometer.GaugeStatisticMeter
+import org.apache.geode.statistics.internal.micrometer.MicrometerMeterGroup
+import org.apache.geode.statistics.internal.micrometer.TimerStatisticMeter
+import org.apache.geode.statistics.util.NOW_NANOS
+
+class DLockStats(private val statMeter: Long) : MicrometerMeterGroup("DLockStats-$statMeter"), DistributedLockStats {
+
+ override fun getCommonTags(): Array<String> = arrayOf("statsMeter", statMeter.toString())
+
+ override fun initializeStaticMeters() {
+ registerMeter(dlockGrantorsMeter)
+ registerMeter(dlockServicesMeter)
+ registerMeter(dlockTokensMeter)
+ registerMeter(dlockGrantTokenMeter)
+ registerMeter(dlockRequestQueuesMeter)
+ registerMeter(dlockSerialQueueSizeMeter)
+ registerMeter(dlockSerialThreadsMeter)
+ registerMeter(dlockWaitingQueueSizeMeter)
+ registerMeter(dlockWaitingThreadsMeter)
+ registerMeter(dlockLockWaitsInProgressMeter)
+ registerMeter(dlockLockWaitsCompletedMeter)
+ registerMeter(dlockLockWaitTimer)
+ registerMeter(dlockLockWaitsFailedMeter)
+ registerMeter(dlockLockWaitsFailedTimer)
+ registerMeter(dlockGrantWaitsInProgressMeter)
+ registerMeter(dlockGrantWaitsCompletedMeter)
+ registerMeter(dlockGrantWaitsTimer)
+ registerMeter(dlockGrantWaitsNotGrantorMeter)
+ registerMeter(dlockGrantWaitsNotGrantorTimer)
+ registerMeter(dlockGrantWaitsTimeoutMeter)
+ registerMeter(dlockGrantWaitsTimeoutTimer)
+ registerMeter(dlockGrantWaitsNotHolderMeter)
+ registerMeter(dlockGrantWaitsNotHolderTimer)
+ registerMeter(dlockGrantWaitsFailedMeter)
+ registerMeter(dlockGrantWaitsFailedTimer)
+ registerMeter(dlockGrantWaitsSuspendedMeter)
+ registerMeter(dlockGrantWaitsSuspendedTimer)
+ registerMeter(dlockGrantWaitsDestroyedMeter)
+ registerMeter(dlockGrantWaitsDestroyedTimer)
+ registerMeter(dlockCreateGrantorInProgressMeter)
+ registerMeter(dlockCreateGrantorCompletedMeter)
+ registerMeter(dlockCreateGrantorTimer)
+ registerMeter(dlockServiceCreatesInProgressMeter)
+ registerMeter(dlockServiceCreatesCompletedMeter)
+ registerMeter(dlockServiceCreatesTimer)
+ registerMeter(dlockServiceInitLatchTimer)
+ registerMeter(dlockGrantorWaitsInProgressMeter)
+ registerMeter(dlockGrantorWaitsCompletedMeter)
+ registerMeter(dlockGrantorWaitsTimer)
+ registerMeter(dlockGrantorWaitsFailedMeter)
+ registerMeter(dlockGrantorWaitsFailedTimer)
+ registerMeter(dlockGrantorThreadsInProgressMeter)
+ registerMeter(dlockGrantorThreadsCompletedMeter)
+ registerMeter(dlockGrantorThreadExpireAndGrantLocksTimer)
+ registerMeter(dlockGrantorThreadHandleRequestTimeoutsTimer)
+ registerMeter(dlockGrantorThreadRemoveUnusedTokensTimer)
+ registerMeter(dlockGrantorThreadTimer)
+ registerMeter(dlockPendingRequestsMeter)
+ registerMeter(dlockDestroyReadWaitsInProgressMeter)
+ registerMeter(dlockDestroyReadWaitsCompletedMeter)
+ registerMeter(dlockDestroyReadWaitsTimer)
+ registerMeter(dlockDestroyReadWaitsFailedMeter)
+ registerMeter(dlockDestroyReadWaitsFailedTimer)
+ registerMeter(dlockDestroyWriteWaitsInProgressMeter)
+ registerMeter(dlockDestroyWriteWaitsCompletedMeter)
+ registerMeter(dlockDestroyWriteWaitsTimer)
+ registerMeter(dlockDestroyWriteWaitsFailedMeter)
+ registerMeter(dlockDestroyWriteWaitsFailerTimer)
+ registerMeter(dlockDestroyReadsMeter)
+ registerMeter(dlockDestroyWritesMeter)
+ registerMeter(dlockLockReleasesInProgressMeter)
+ registerMeter(dlockLockReleasesCompletedMeter)
+ registerMeter(dlockLockReleasesTimer)
+ registerMeter(dlockBecomeGrantorRequestsMeter)
+ registerMeter(dlockFreeResourcesCompleteMeter)
+ registerMeter(dlockFreeResourcesFailedMeter)
+ }
+
+ private val dlockGrantorsMeter = GaugeStatisticMeter("dlock.grantors", "The current number of lock grantors hosted by this system member.")
+ private val dlockServicesMeter = GaugeStatisticMeter("dlock.services", "The current number of lock services used by this system member.")
+ private val dlockTokensMeter = GaugeStatisticMeter("dlock.tokens", "The current number of lock tokens used by this system member.")
+ private val dlockGrantTokenMeter = GaugeStatisticMeter("dlock.grantTokens", "The current number of grant tokens used by local grantors.")
+ private val dlockRequestQueuesMeter = GaugeStatisticMeter("dlock.requestQueues", "The current number of lock request queues used by this system member.")
+ private val dlockSerialQueueSizeMeter = GaugeStatisticMeter("dlock.serialQueueSize", "The number of serial distribution messages currently waiting to be processed.")
+ private val dlockSerialThreadsMeter = GaugeStatisticMeter("dlock.serialThreads", "The number of threads currently processing serial/ordered messages.")
+ private val dlockWaitingQueueSizeMeter = GaugeStatisticMeter("dlock.waitingQueueSize", "The number of distribution messages currently waiting for some other resource before they can be processed.")
+ private val dlockWaitingThreadsMeter = GaugeStatisticMeter("dlock.waitingThreads", "The number of threads currently processing messages that had to wait for a resource.")
+ private val dlockLockWaitsInProgressMeter = GaugeStatisticMeter("dlock.lockWaitsInProgress", "Current number of threads waiting for a distributed lock.")
+ private val dlockLockWaitsCompletedMeter = CounterStatisticMeter("dlock.lockWaitsCompleted", "Total number of times distributed lock wait has completed by successfully obtained the lock.")
+ private val dlockLockWaitTimer = TimerStatisticMeter("dlock.lockWaitTime", "Total time spent waiting for a distributed lock that was obtained.", unit = "nanoseconds")
+ private val dlockLockWaitsFailedMeter = CounterStatisticMeter("dlock.lockWaitsFailed", "Total number of times distributed lock wait has completed by failing to obtain the lock.")
+ private val dlockLockWaitsFailedTimer = TimerStatisticMeter("dlock.lockWaitFailedTime", "Total time spent waiting for a distributed lock that we failed to obtain.", unit = "nanoseconds")
+ private val dlockGrantWaitsInProgressMeter = GaugeStatisticMeter("dlock.grantWaitsInProgress", "Current number of distributed lock requests being granted.")
+ private val dlockGrantWaitsCompletedMeter = CounterStatisticMeter("dlock.grantWaitsCompleted", "Total number of times granting of a lock request has completed by successfully granting the lock.")
+ private val dlockGrantWaitsTimer = TimerStatisticMeter("dlock.grantWaitTime", "Total time spent attempting to grant a distributed lock.", unit = "nanoseconds")
+ private val dlockGrantWaitsNotGrantorMeter = CounterStatisticMeter("dlock.grantWaitsNotGrantor", "Total number of times granting of lock request failed because not grantor.")
+ private val dlockGrantWaitsNotGrantorTimer = TimerStatisticMeter("dlock.grantWaitNotGrantorTime", "Total time spent granting of lock requests that failed because not grantor.", unit = "nanoseconds")
+ private val dlockGrantWaitsTimeoutMeter = CounterStatisticMeter("dlock.grantWaitsTimeout", "Total number of times granting of lock request failed because timeout.")
+ private val dlockGrantWaitsTimeoutTimer = TimerStatisticMeter("grantWaitTimeoutTime", "Total time spent granting of lock requests that failed because timeout.", unit = "nanoseconds")
+ private val dlockGrantWaitsNotHolderMeter = CounterStatisticMeter("dlock.grantWaitsNotHolder", "Total number of times granting of lock request failed because reentrant was not holder.")
+ private val dlockGrantWaitsNotHolderTimer = TimerStatisticMeter("dlock.grantWaitNotHolderTime", "Total time spent granting of lock requests that failed because reentrant was not holder.", unit = "nanoseconds")
+ private val dlockGrantWaitsFailedMeter = CounterStatisticMeter("dlock.grantWaitsFailed", "Total number of times granting of lock request failed because try locks failed.")
+ private val dlockGrantWaitsFailedTimer = TimerStatisticMeter("dlock.grantWaitFailedTime", "Total time spent granting of lock requests that failed because try locks failed.", unit = "nanoseconds")
+ private val dlockGrantWaitsSuspendedMeter = CounterStatisticMeter("dlock.grantWaitsSuspended", "Total number of times granting of lock request failed because lock service was suspended.")
+ private val dlockGrantWaitsSuspendedTimer = TimerStatisticMeter("dlock.grantWaitSuspendedTime", "Total time spent granting of lock requests that failed because lock service was suspended.", unit = "nanoseconds")
+ private val dlockGrantWaitsDestroyedMeter = CounterStatisticMeter("dlock.grantWaitsDestroyed", "Total number of times granting of lock request failed because lock service was destroyed.")
+ private val dlockGrantWaitsDestroyedTimer = TimerStatisticMeter("dlock.grantWaitDestroyedTime", "Total time spent granting of lock requests that failed because lock service was destroyed.", unit = "nanoseconds")
+ private val dlockCreateGrantorInProgressMeter = GaugeStatisticMeter("dlock.createGrantorsInProgress", "Current number of initial grantors being created in this process.")
+ private val dlockCreateGrantorCompletedMeter = CounterStatisticMeter("dlock.createGrantorsCompleted", "Total number of initial grantors created in this process.")
+ private val dlockCreateGrantorTimer = TimerStatisticMeter("dlock.createGrantorTime", "Total time spent waiting create the intial grantor for lock services.", unit = "nanoseconds")
+ private val dlockServiceCreatesInProgressMeter = GaugeStatisticMeter("dlock.serviceCreatesInProgress", "Current number of lock services being created in this process.")
+ private val dlockServiceCreatesCompletedMeter = CounterStatisticMeter("dlock.serviceCreatesCompleted", "Total number of lock services created in this process.")
+ private val dlockServiceCreatesTimer = TimerStatisticMeter("dlock.serviceCreateLatchTime", "Total time spent creating lock services before releasing create latches.", unit = "nanoseconds")
+ private val dlockServiceInitLatchTimer = TimerStatisticMeter("dlock.serviceInitLatchTime", "Total time spent creating lock services before releasing init latches.", unit = "nanoseconds")
+ private val dlockGrantorWaitsInProgressMeter = GaugeStatisticMeter("dlock.grantorWaitsInProgress", "Current number of threads waiting for grantor latch to open.")
+ private val dlockGrantorWaitsCompletedMeter = CounterStatisticMeter("dlock.grantorWaitsCompleted", "Total number of times waiting threads completed waiting for the grantor latch to open.")
+ private val dlockGrantorWaitsTimer = TimerStatisticMeter("dlock.grantorWaitTime", "Total time spent waiting for the grantor latch which resulted in success.", unit = "nanoseconds")
+ private val dlockGrantorWaitsFailedMeter = CounterStatisticMeter("dlock.grantorWaitsFailed", "Total number of times waiting threads failed to finish waiting for the grantor latch to open.")
+ private val dlockGrantorWaitsFailedTimer = TimerStatisticMeter("dlock.grantorWaitFailedTime", "Total time spent waiting for the grantor latch which resulted in failure.", unit = "nanoseconds")
+ private val dlockGrantorThreadsInProgressMeter = GaugeStatisticMeter("dlock.grantorThreadsInProgress", "Current iterations of work performed by grantor thread(s).")
+ private val dlockGrantorThreadsCompletedMeter = CounterStatisticMeter("dlock.grantorThreadsCompleted", "Total number of iterations of work performed by grantor thread(s).")
+ private val dlockGrantorThreadExpireAndGrantLocksTimer = TimerStatisticMeter("dlock.grantorThreadExpireAndGrantLocksTime", "Total time spent by grantor thread(s) performing expireAndGrantLocks tasks.", unit = "nanoseconds")
+ private val dlockGrantorThreadHandleRequestTimeoutsTimer = TimerStatisticMeter("dlock.grantorThreadHandleRequestTimeoutsTime", "Total time spent by grantor thread(s) performing handleRequestTimeouts tasks.", unit = "nanoseconds")
+ private val dlockGrantorThreadRemoveUnusedTokensTimer = TimerStatisticMeter("dlock.grantorThreadRemoveUnusedTokensTime", "Total time spent by grantor thread(s) performing removeUnusedTokens tasks.", unit = "nanoseconds")
+ private val dlockGrantorThreadTimer = TimerStatisticMeter("dlock.grantorThreadTime", "Total time spent by grantor thread(s) performing all grantor tasks.", unit = "nanoseconds")
+ private val dlockPendingRequestsMeter = GaugeStatisticMeter("dlock.pendingRequests", "The current number of pending lock requests queued by grantors in this process.")
+ private val dlockDestroyReadWaitsInProgressMeter = GaugeStatisticMeter("dlock.destroyReadWaitsInProgress", "Current number of threads waiting for a DLockService destroy read lock.")
+ private val dlockDestroyReadWaitsCompletedMeter = CounterStatisticMeter("dlock.destroyReadWaitsCompleted", "Total number of times a DLockService destroy read lock wait has completed successfully.")
+ private val dlockDestroyReadWaitsTimer = TimerStatisticMeter("dlock.destroyReadWaitTime", "Total time spent waiting for a DLockService destroy read lock that was obtained.", unit = "nanoseconds")
+ private val dlockDestroyReadWaitsFailedMeter = CounterStatisticMeter("dlock.destroyReadWaitsFailed", "Total number of times a DLockService destroy read lock wait has completed unsuccessfully.")
+ private val dlockDestroyReadWaitsFailedTimer = TimerStatisticMeter("dlock.destroyReadWaitFailedTime", "Total time spent waiting for a DLockService destroy read lock that was not obtained.", unit = "nanoseconds")
+ private val dlockDestroyWriteWaitsInProgressMeter = GaugeStatisticMeter("dlock.destroyWriteWaitsInProgress", "Current number of thwrites waiting for a DLockService destroy write lock.")
+ private val dlockDestroyWriteWaitsCompletedMeter = CounterStatisticMeter("dlock.destroyWriteWaitsCompleted", "Total number of times a DLockService destroy write lock wait has completed successfully.")
+ private val dlockDestroyWriteWaitsTimer = TimerStatisticMeter("dlock.destroyWriteWaitTime", "Total time spent waiting for a DLockService destroy write lock that was obtained.", unit = "nanoseconds")
+ private val dlockDestroyWriteWaitsFailedMeter = CounterStatisticMeter("dlock.destroyWriteWaitsFailed", "Total number of times a DLockService destroy write lock wait has completed unsuccessfully.")
+ private val dlockDestroyWriteWaitsFailerTimer = TimerStatisticMeter("dlock.destroyWriteWaitFailedTime", "Total time spent waiting for a DLockService destroy write lock that was not obtained.", unit = "nanoseconds")
+ private val dlockDestroyReadsMeter = GaugeStatisticMeter("dlock.destroyReads", "The current number of DLockService destroy read locks held by this process.")
+ private val dlockDestroyWritesMeter = GaugeStatisticMeter("dlock.destroyWrites", "The current number of DLockService destroy write locks held by this process.")
+ private val dlockLockReleasesInProgressMeter = GaugeStatisticMeter("dlock.lockReleasesInProgress", "Current number of threads releasing a distributed lock.")
+ private val dlockLockReleasesCompletedMeter = CounterStatisticMeter("dlock.lockReleasesCompleted", "Total number of times distributed lock release has completed.")
+ private val dlockLockReleasesTimer = TimerStatisticMeter("dlock.lockReleaseTime", "Total time spent releasing a distributed lock.", unit = "nanoseconds")
+ private val dlockBecomeGrantorRequestsMeter = CounterStatisticMeter("dlock.becomeGrantorRequests", "Total number of times this member has explicitly requested to become lock grantor.")
+ private val dlockFreeResourcesCompleteMeter = CounterStatisticMeter("dlock.freeResourcesCompleted", "Total number of times this member has freed resources for a distributed lock.")
+ private val dlockFreeResourcesFailedMeter = CounterStatisticMeter("dlock.freeResourcesFailed", "Total number of times this member has attempted to free resources for a distributed lock which remained in use.")
+
+
+ val grantWaitsSuspended: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = dlockGrantWaitsSuspendedMeter.getValue()
+
+ val grantWaitsDestroyed: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = dlockGrantWaitsDestroyedMeter.getValue()
+
+ override fun getLockWaitsInProgress(): Long = dlockLockWaitsInProgressMeter.getValue()
+ override fun getLockWaitsCompleted(): Long = dlockLockWaitsCompletedMeter.getValue()
+ override fun getLockWaitsFailed(): Long = dlockLockWaitsFailedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getLockWaitTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getLockWaitFailedTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ override fun startLockWait(): Long {
+ dlockLockWaitsInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun endLockWait(start: Long, success: Boolean) {
+ val ts = NOW_NANOS
+ dlockLockWaitsInProgressMeter.decrement()
+ if (success) {
+ dlockLockWaitsCompletedMeter.increment()
+ dlockLockWaitTimer.recordValue(ts - start)
+ } else {
+ dlockLockWaitsFailedMeter.increment()
+ dlockLockWaitsFailedTimer.recordValue(ts - start)
+ }
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getWaitingQueueSize(): Long = dlockWaitingQueueSizeMeter.getValue()
+
+ override fun incWaitingQueueSize(messages: Long) {
+ dlockWaitingQueueSizeMeter.increment(messages)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getSerialQueueSize(): Long = dlockSerialQueueSizeMeter.getValue()
+
+ override fun incSerialQueueSize(messages: Long) {
+ dlockSerialQueueSizeMeter.increment(messages)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getNumSerialThreads(): Long = dlockSerialThreadsMeter.getValue()
+
+ override fun incNumSerialThreads(threads: Long) {
+ dlockSerialThreadsMeter.increment(threads)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getWaitingThreads(): Long = dlockWaitingThreadsMeter.getValue()
+
+ override fun incWaitingThreads(threads: Long) {
+ dlockWaitingThreadsMeter.increment(threads)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getServices(): Long = dlockServicesMeter.getValue()
+
+ override fun incServices(value: Long) {
+ dlockServicesMeter.increment(value)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantors(): Long = dlockGrantorsMeter.getValue()
+
+ override fun incGrantors(value: Long) {
+ dlockGrantorsMeter.increment(value)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getTokens(): Long = dlockTokensMeter.getValue()
+
+ override fun incTokens(value: Long) {
+ dlockTokensMeter.increment(value)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantTokens(): Long = dlockGrantTokenMeter.getValue()
+
+ override fun incGrantTokens(value: Long) {
+ dlockGrantTokenMeter.increment(value)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getRequestQueues(): Long = dlockRequestQueuesMeter.getValue()
+
+ override fun incRequestQueues(value: Long) {
+ dlockRequestQueuesMeter.increment(value)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantWaitsInProgress(): Long = dlockGrantWaitsInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantWaitsCompleted(): Long = dlockGrantWaitsInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantWaitsFailed(): Long = dlockGrantWaitsFailedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantWaitTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantWaitFailedTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ override fun startGrantWait(): Long {
+ dlockGrantWaitsInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun endGrantWait(start: Long) {
+ dlockGrantWaitsTimer.recordValue(NOW_NANOS - start)
+ dlockGrantWaitsInProgressMeter.decrement()
+ dlockGrantWaitsCompletedMeter.increment()
+ }
+
+ override fun endGrantWaitNotGrantor(start: Long) {
+ dlockGrantWaitsNotGrantorTimer.recordValue(NOW_NANOS - start)
+ dlockGrantWaitsInProgressMeter.decrement()
+ dlockGrantWaitsNotGrantorMeter.increment()
+ }
+
+ override fun endGrantWaitTimeout(start: Long) {
+ dlockGrantWaitsTimeoutTimer.recordValue(NOW_NANOS - start)
+ dlockGrantWaitsInProgressMeter.decrement()
+ dlockGrantWaitsTimeoutMeter.increment()
+ }
+
+ override fun endGrantWaitNotHolder(start: Long) {
+ dlockGrantWaitsNotHolderTimer.recordValue(NOW_NANOS - start)
+ dlockGrantWaitsInProgressMeter.decrement()
+ dlockGrantWaitsNotHolderMeter.increment()
+ }
+
+ override fun endGrantWaitFailed(start: Long) {
+ dlockGrantWaitsFailedTimer.recordValue(NOW_NANOS - start)
+ dlockGrantWaitsInProgressMeter.decrement()
+ dlockGrantWaitsFailedMeter.increment()
+ }
+
+ override fun endGrantWaitSuspended(start: Long) {
+ dlockGrantWaitsSuspendedTimer.recordValue(NOW_NANOS - start)
+ dlockGrantWaitsInProgressMeter.decrement()
+ dlockGrantWaitsSuspendedMeter.increment()
+ }
+
+ override fun endGrantWaitDestroyed(start: Long) {
+ dlockGrantWaitsDestroyedTimer.recordValue(NOW_NANOS - start)
+ dlockGrantWaitsInProgressMeter.decrement()
+ dlockGrantWaitsDestroyedMeter.increment()
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getCreateGrantorsInProgress(): Long = dlockCreateGrantorInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getCreateGrantorsCompleted(): Long = dlockCreateGrantorCompletedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getCreateGrantorTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ override fun startCreateGrantor(): Long {
+ dlockCreateGrantorInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun endCreateGrantor(start: Long) {
+ dlockCreateGrantorTimer.recordValue(NOW_NANOS - start)
+ dlockCreateGrantorInProgressMeter.decrement()
+ dlockCreateGrantorCompletedMeter.increment()
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getServiceCreatesInProgress(): Long = dlockServiceCreatesInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getServiceCreatesCompleted(): Long = dlockServiceCreatesCompletedMeter.getValue()
+
+ override fun startServiceCreate(): Long {
+ dlockServiceCreatesInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun serviceCreateLatchReleased(start: Long) {
+ dlockServiceInitLatchTimer.recordValue(NOW_NANOS - start)
+ }
+
+ override fun serviceInitLatchReleased(start: Long) {
+ dlockServiceInitLatchTimer.recordValue(NOW_NANOS - start)
+ dlockServiceCreatesInProgressMeter.decrement()
+ dlockServiceCreatesCompletedMeter.increment()
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getServiceCreateLatchTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getServiceInitLatchTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorWaitsInProgress(): Long = dlockGrantorWaitsInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorWaitsCompleted(): Long = dlockGrantorWaitsCompletedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorWaitsFailed(): Long = dlockGrantorWaitsFailedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorWaitTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorWaitFailedTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ override fun startGrantorWait(): Long {
+ dlockGrantorWaitsInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun endGrantorWait(start: Long, success: Boolean) {
+ dlockGrantorWaitsInProgressMeter.decrement()
+ if (success) {
+ dlockGrantorWaitsCompletedMeter.increment()
+ dlockGrantorWaitsTimer.recordValue(NOW_NANOS - start)
+ } else {
+ dlockGrantorWaitsFailedMeter.increment()
+ dlockGrantorWaitsFailedTimer.recordValue(NOW_NANOS - start)
+ }
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorThreadsInProgress(): Long = dlockGrantorThreadsInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorThreadsCompleted(): Long = dlockGrantorThreadsCompletedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorThreadTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorThreadExpireAndGrantLocksTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorThreadHandleRequestTimeoutsTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getGrantorThreadRemoveUnusedTokensTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ override fun startGrantorThread(): Long {
+ dlockGrantorThreadsInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun endGrantorThreadExpireAndGrantLocks(start: Long): Long {
+ dlockGrantorThreadExpireAndGrantLocksTimer.recordValue(NOW_NANOS - start)
+ return NOW_NANOS
+ }
+
+ override fun endGrantorThreadHandleRequestTimeouts(timing: Long): Long {
+ dlockGrantorThreadHandleRequestTimeoutsTimer.recordValue(NOW_NANOS - timing)
+ return NOW_NANOS
+ }
+
+ override fun endGrantorThreadRemoveUnusedTokens(timing: Long) {
+ dlockGrantorThreadRemoveUnusedTokensTimer.recordValue(NOW_NANOS - timing)
+ }
+
+ override fun endGrantorThread(start: Long) {
+ dlockGrantorThreadsInProgressMeter.decrement()
+ dlockGrantorThreadsCompletedMeter.increment()
+ dlockGrantorThreadTimer.recordValue(NOW_NANOS - start)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getPendingRequests(): Long = dlockPendingRequestsMeter.getValue()
+
+ override fun incPendingRequests(value: Long) {
+ dlockPendingRequestsMeter.increment(value)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyReadWaitsInProgress(): Long = dlockDestroyReadWaitsInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyReadWaitsCompleted(): Long = dlockDestroyReadWaitsCompletedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyReadWaitsFailed(): Long = dlockDestroyReadWaitsFailedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyReadWaitTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyReadWaitFailedTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ override fun startDestroyReadWait(): Long {
+ dlockDestroyReadWaitsInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun endDestroyReadWait(start: Long, success: Boolean) {
+ dlockDestroyReadWaitsInProgressMeter.decrement()
+ if (success) {
+ dlockDestroyReadWaitsCompletedMeter.increment()
+ dlockDestroyReadWaitsTimer.recordValue(NOW_NANOS - start)
+ } else {
+ dlockDestroyReadWaitsFailedMeter.increment()
+ dlockDestroyReadWaitsFailedTimer.recordValue(NOW_NANOS - start)
+ }
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyWriteWaitsInProgress(): Long = dlockDestroyWriteWaitsInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyWriteWaitsCompleted(): Long = dlockDestroyWriteWaitsCompletedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyWriteWaitsFailed(): Long = dlockDestroyWriteWaitsFailedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyWriteWaitTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyWriteWaitFailedTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ override fun startDestroyWriteWait(): Long {
+ dlockDestroyWriteWaitsInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun endDestroyWriteWait(start: Long, success: Boolean) {
+ dlockDestroyWriteWaitsInProgressMeter.decrement()
+ if (success) {
+ dlockDestroyWriteWaitsCompletedMeter.increment()
+ dlockDestroyWriteWaitsTimer.recordValue(NOW_NANOS - start)
+ } else {
+ dlockDestroyWriteWaitsFailedMeter.increment()
+ dlockDestroyWriteWaitsFailerTimer.recordValue(NOW_NANOS - start)
+ }
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyReads(): Long = dlockDestroyReadsMeter.getValue()
+
+ override fun incDestroyReads(value: Long) {
+ dlockDestroyReadsMeter.increment(value)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getDestroyWrites(): Long = dlockDestroyWritesMeter.getValue()
+
+ override fun incDestroyWrites(value: Long) {
+ dlockDestroyWritesMeter.increment(value)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getLockReleasesInProgress(): Long = dlockLockReleasesInProgressMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getLockReleasesCompleted(): Long = dlockLockReleasesCompletedMeter.getValue()
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getLockReleaseTime(): Long {
+ TODO("We should not expose the timer stats publically")
+ }
+
+ override fun startLockRelease(): Long {
+ dlockLockReleasesInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ override fun endLockRelease(start: Long) {
+ dlockLockReleasesInProgressMeter.decrement()
+ dlockLockReleasesCompletedMeter.increment()
+ dlockLockReleasesTimer.recordValue(NOW_NANOS - start)
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getBecomeGrantorRequests(): Long = dlockBecomeGrantorRequestsMeter.getValue()
+
+ override fun incBecomeGrantorRequests() {
+ dlockBecomeGrantorRequestsMeter.increment()
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getFreeResourcesCompleted(): Long = dlockFreeResourcesCompleteMeter.getValue()
+
+ override fun incFreeResourcesCompleted() {
+ dlockFreeResourcesCompleteMeter.increment()
+ }
+
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ override fun getFreeResourcesFailed(): Long = dlockFreeResourcesFailedMeter.getValue()
+
+ override fun incFreeResourcesFailed() {
+ dlockFreeResourcesFailedMeter.increment()
+ }
+
+ override fun getSerialQueueHelper(): QueueStatHelper {
+ return object : QueueStatHelper {
+ override fun add() {
+ incSerialQueueSize(1)
+ }
+
+ override fun remove() {
+ incSerialQueueSize(-1)
+ }
+
+ override fun remove(count: Int) {
+ incSerialQueueSize(-count.toLong())
+ }
+ }
+ }
+
+ override fun getWaitingPoolHelper(): PoolStatHelper {
+ return object : PoolStatHelper {
+ override fun startJob() {
+ incWaitingThreads(1)
+ }
+
+ override fun endJob() {
+ incWaitingThreads(-1)
+ }
+ }
+ }
+
+ override fun getWaitingQueueHelper(): QueueStatHelper {
+ return object : QueueStatHelper {
+ override fun add() {
+ incWaitingQueueSize(1)
+ }
+
+ override fun remove() {
+ incWaitingQueueSize(-1)
+ }
+
+ override fun remove(count: Int) {
+ incWaitingQueueSize(-count.toLong())
+ }
+ }
+ }
+}
diff --git a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/dlock/DistributedLockStats.java b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/dlock/DistributedLockStats.java
new file mode 100755
index 0000000..bdab589
--- /dev/null
+++ b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/dlock/DistributedLockStats.java
@@ -0,0 +1,270 @@
+/*
+ * 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.statistics.dlock;
+
+import org.apache.geode.distributed.internal.PoolStatHelper;
+import org.apache.geode.distributed.internal.QueueStatHelper;
+
+/**
+ * Defines the interface used to access and modify distributed lock statistics.
+ *
+ *
+ */
+public interface DistributedLockStats {
+
+ // time for call to lock() to complete
+
+ /**
+ * Returns the number of threads currently waiting for a distributed lock
+ */
+ long getLockWaitsInProgress();
+
+ /**
+ * Returns the total number of waits for a distributed lock
+ */
+ long getLockWaitsCompleted();
+
+ long getLockWaitsFailed();
+
+ /**
+ * Returns the total number of nanoseconds spent waiting for a distributed lock.
+ */
+ long getLockWaitTime();
+
+ long getLockWaitFailedTime();
+
+ /**
+ * @return the timestamp that marks the start of the operation
+ */
+ long startLockWait();
+
+ /**
+ * @param start the timestamp taken when the operation started
+ */
+ void endLockWait(long start, boolean success);
+
+ // incSerialQueueSize everytime getWaitingQueueHelper add/remove called
+ long getWaitingQueueSize();
+
+ void incWaitingQueueSize(long messages);
+
+ // incSerialQueueSize everytime getSerialQueueHelper add/remove called
+ long getSerialQueueSize();
+
+ void incSerialQueueSize(long messages);
+
+ // incNumSerialThreads everytime we execute with dlock getSerialExecutor()
+ long getNumSerialThreads();
+
+ void incNumSerialThreads(long threads);
+
+ // incWaitingThreads for every invoke of getWaitingPoolHelper startJob/endJob
+ long getWaitingThreads();
+
+ void incWaitingThreads(long threads);
+
+ // current number of lock services used by this system member
+ long getServices();
+
+ void incServices(long val);
+
+ // current number of lock grantors hosted by this system member
+ long getGrantors();
+
+ void incGrantors(long val);
+
+ // time spent granting of lock requests to completion
+ long getGrantWaitsInProgress();
+
+ long getGrantWaitsCompleted();
+
+ long getGrantWaitsFailed();
+
+ long getGrantWaitTime();
+
+ long getGrantWaitFailedTime();
+
+ long startGrantWait();
+
+ void endGrantWait(long start);
+
+ void endGrantWaitNotGrantor(long start);
+
+ void endGrantWaitTimeout(long start);
+
+ void endGrantWaitNotHolder(long start);
+
+ void endGrantWaitFailed(long start);
+
+ void endGrantWaitDestroyed(long start);
+
+ void endGrantWaitSuspended(long start);
+
+ // time spent creating initial grantor for lock service
+ long getCreateGrantorsInProgress();
+
+ long getCreateGrantorsCompleted();
+
+ long getCreateGrantorTime();
+
+ long startCreateGrantor();
+
+ void endCreateGrantor(long start);
+
+ // time spent creating each lock service
+ long getServiceCreatesInProgress();
+
+ long getServiceCreatesCompleted();
+
+ long startServiceCreate();
+
+ void serviceCreateLatchReleased(long start);
+
+ void serviceInitLatchReleased(long start);
+
+ long getServiceCreateLatchTime();
+
+ long getServiceInitLatchTime();
+
+ // time spent waiting for grantor latches to open
+ long getGrantorWaitsInProgress();
+
+ long getGrantorWaitsCompleted();
+
+ long getGrantorWaitsFailed();
+
+ long getGrantorWaitTime();
+
+ long getGrantorWaitFailedTime();
+
+ long startGrantorWait();
+
+ void endGrantorWait(long start, boolean success);
+
+ // helpers for executor usage
+ QueueStatHelper getSerialQueueHelper();
+
+ PoolStatHelper getWaitingPoolHelper();
+
+ QueueStatHelper getWaitingQueueHelper();
+
+ // time spent by grantor threads
+ long getGrantorThreadsInProgress();
+
+ long getGrantorThreadsCompleted();
+
+ long getGrantorThreadTime();
+
+ long getGrantorThreadExpireAndGrantLocksTime();
+
+ long getGrantorThreadHandleRequestTimeoutsTime();
+
+ long getGrantorThreadRemoveUnusedTokensTime();
+
+ long startGrantorThread();
+
+ long endGrantorThreadExpireAndGrantLocks(long start);
+
+ long endGrantorThreadHandleRequestTimeouts(long timing);
+
+ void endGrantorThreadRemoveUnusedTokens(long timing);
+
+ void endGrantorThread(long start);
+
+ // current number of lock grantors hosted by this system member
+ long getPendingRequests();
+
+ void incPendingRequests(long val);
+
+ // acquisition of destroyReadLock in DLockService
+ long getDestroyReadWaitsInProgress();
+
+ long getDestroyReadWaitsCompleted();
+
+ long getDestroyReadWaitsFailed();
+
+ long getDestroyReadWaitTime();
+
+ long getDestroyReadWaitFailedTime();
+
+ long startDestroyReadWait();
+
+ void endDestroyReadWait(long start, boolean success);
+
+ // acquisition of destroyWriteLock in DLockService
+ long getDestroyWriteWaitsInProgress();
+
+ long getDestroyWriteWaitsCompleted();
+
+ long getDestroyWriteWaitsFailed();
+
+ long getDestroyWriteWaitTime();
+
+ long getDestroyWriteWaitFailedTime();
+
+ long startDestroyWriteWait();
+
+ void endDestroyWriteWait(long start, boolean success);
+
+ // current number of DLockService destroy read locks held by this process
+ long getDestroyReads();
+
+ void incDestroyReads(long val);
+
+ // current number of DLockService destroy write locks held by this process
+ long getDestroyWrites();
+
+ void incDestroyWrites(long val);
+
+ // time for call to unlock() to complete
+ long getLockReleasesInProgress();
+
+ long getLockReleasesCompleted();
+
+ long getLockReleaseTime();
+
+ long startLockRelease();
+
+ void endLockRelease(long start);
+
+ // total number of times this member has requested to become grantor
+ long getBecomeGrantorRequests();
+
+ void incBecomeGrantorRequests();
+
+ // current number of lock tokens used by this system member
+ long getTokens();
+
+ void incTokens(long val);
+
+ // current number of grant tokens used by local grantors
+ long getGrantTokens();
+
+ void incGrantTokens(long val);
+
+ // current number of lock request queues used by this system member
+ long getRequestQueues();
+
+ void incRequestQueues(long val);
+
+ long getFreeResourcesCompleted();
+
+ void incFreeResourcesCompleted();
+
+ long getFreeResourcesFailed();
+
+ void incFreeResourcesFailed();
+}
diff --git a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/AsyncEventQueueStats.java b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/AsyncEventQueueStats.java
new file mode 100644
index 0000000..746dd1b
--- /dev/null
+++ b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/AsyncEventQueueStats.java
@@ -0,0 +1,186 @@
+/*
+ * 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.statistics.wan;
+
+import org.apache.geode.statistics.StatisticDescriptor;
+import org.apache.geode.statistics.StatisticsFactory;
+import org.apache.geode.statistics.StatisticsType;
+import org.apache.geode.statistics.StatisticsTypeFactory;
+import org.apache.geode.internal.cache.wan.GatewaySenderStats;
+import org.apache.geode.internal.statistics.StatisticsTypeFactoryImpl;
+
+public class AsyncEventQueueStats extends GatewaySenderStats {
+
+ public static final String typeName = "AsyncEventQueueStatistics";
+
+ /**
+ * The <code>StatisticsType</code> of the statistics
+ */
+ private StatisticsType type;
+
+ @Override
+ protected void initializeStats(StatisticsFactory factory) {
+ type = factory.createType(typeName, "Stats for activity in the AsyncEventQueue",
+ new StatisticDescriptor[]{
+ factory.createIntCounter(GatewaySenderStats.EVENTS_RECEIVED, "Number of events received by this queue.",
+ "operations"),
+ factory.createIntCounter(GatewaySenderStats.EVENTS_QUEUED, "Number of events added to the event queue.",
+ "operations"),
+ factory.createLongCounter(GatewaySenderStats.EVENT_QUEUE_TIME, "Total time spent queueing events.",
+ "nanoseconds"),
+ factory.createIntGauge(GatewaySenderStats.EVENT_QUEUE_SIZE, "Size of the event queue.", "operations",
+ false),
+ factory.createIntGauge(GatewaySenderStats.SECONDARY_EVENT_QUEUE_SIZE, "Size of the secondary event queue.",
+ "operations", false),
+ factory.createIntGauge(GatewaySenderStats.EVENTS_PROCESSED_BY_PQRM,
+ "Total number of events processed by Parallel Queue Removal Message(PQRM).",
+ "operations", false),
+ factory.createIntGauge(GatewaySenderStats.TMP_EVENT_QUEUE_SIZE, "Size of the temporary events queue.",
+ "operations", false),
+ factory.createIntCounter(GatewaySenderStats.EVENTS_NOT_QUEUED_CONFLATED,
+ "Number of events received but not added to the event queue because the queue already contains an event with the event's key.",
+ "operations"),
+ factory.createIntCounter(GatewaySenderStats.EVENTS_CONFLATED_FROM_BATCHES,
+ "Number of events conflated from batches.", "operations"),
+ factory.createIntCounter(GatewaySenderStats.EVENTS_DISTRIBUTED,
+ "Number of events removed from the event queue and sent.", "operations"),
+ factory.createIntCounter(GatewaySenderStats.EVENTS_EXCEEDING_ALERT_THRESHOLD,
+ "Number of events exceeding the alert threshold.", "operations", false),
+ factory.createLongCounter(GatewaySenderStats.BATCH_DISTRIBUTION_TIME,
+ "Total time spent distributing batches of events to receivers.", "nanoseconds"),
+ factory.createIntCounter(GatewaySenderStats.BATCHES_DISTRIBUTED,
+ "Number of batches of events removed from the event queue and sent.", "operations"),
+ factory.createIntCounter(GatewaySenderStats.BATCHES_REDISTRIBUTED,
+ "Number of batches of events removed from the event queue and resent.",
+ "operations", false),
+ factory.createIntCounter(GatewaySenderStats.UNPROCESSED_TOKENS_ADDED_BY_PRIMARY,
+ "Number of tokens added to the secondary's unprocessed token map by the primary (though a listener).",
+ "tokens"),
+ factory.createIntCounter(GatewaySenderStats.UNPROCESSED_EVENTS_ADDED_BY_SECONDARY,
+ "Number of events added to the secondary's unprocessed event map by the secondary.",
+ "events"),
+ factory.createIntCounter(GatewaySenderStats.UNPROCESSED_EVENTS_REMOVED_BY_PRIMARY,
+ "Number of events removed from the secondary's unprocessed event map by the primary (though a listener).",
+ "events"),
+ factory.createIntCounter(GatewaySenderStats.UNPROCESSED_TOKENS_REMOVED_BY_SECONDARY,
+ "Number of tokens removed from the secondary's unprocessed token map by the secondary.",
+ "tokens"),
+ factory.createIntCounter(GatewaySenderStats.UNPROCESSED_EVENTS_REMOVED_BY_TIMEOUT,
+ "Number of events removed from the secondary's unprocessed event map by a timeout.",
+ "events"),
+ factory.createIntCounter(GatewaySenderStats.UNPROCESSED_TOKENS_REMOVED_BY_TIMEOUT,
+ "Number of tokens removed from the secondary's unprocessed token map by a timeout.",
+ "tokens"),
+ factory.createIntGauge(GatewaySenderStats.UNPROCESSED_EVENT_MAP_SIZE,
+ "Current number of entries in the secondary's unprocessed event map.", "events",
+ false),
+ factory.createIntGauge(GatewaySenderStats.UNPROCESSED_TOKEN_MAP_SIZE,
+ "Current number of entries in the secondary's unprocessed token map.", "tokens",
+ false),
+ factory.createIntGauge(GatewaySenderStats.CONFLATION_INDEXES_MAP_SIZE,
+ "Current number of entries in the conflation indexes map.", "events"),
+ factory.createIntCounter(GatewaySenderStats.NOT_QUEUED_EVENTS, "Number of events not added to queue.",
+ "events"),
+ factory.createIntCounter(
+ GatewaySenderStats.EVENTS_DROPPED_DUE_TO_PRIMARY_SENDER_NOT_RUNNING,
+ "Number of events dropped because the primary gateway sender is not running.",
+ "events"),
+ factory.createIntCounter(GatewaySenderStats.EVENTS_FILTERED,
+ "Number of events filtered through GatewayEventFilter.", "events"),
+ factory.createIntCounter(GatewaySenderStats.LOAD_BALANCES_COMPLETED, "Number of load balances completed",
+ "operations"),
+ factory.createIntGauge(GatewaySenderStats.LOAD_BALANCES_IN_PROGRESS, "Number of load balances in progress",
+ "operations"),
+ factory.createLongCounter(GatewaySenderStats.LOAD_BALANCE_TIME,
+ "Total time spent load balancing this sender",
+ "nanoseconds"),
+ factory.createIntCounter(GatewaySenderStats.SYNCHRONIZATION_EVENTS_ENQUEUED,
+ "Number of synchronization events added to the event queue.", "operations"),
+ factory.createIntCounter(GatewaySenderStats.SYNCHRONIZATION_EVENTS_PROVIDED,
+ "Number of synchronization events provided to other members.", "operations"),});
+
+ // Initialize id fields
+ GatewaySenderStats.eventsReceivedId = type.nameToId(GatewaySenderStats.EVENTS_RECEIVED);
+ GatewaySenderStats.eventsQueuedId = type.nameToId(GatewaySenderStats.EVENTS_QUEUED);
+ GatewaySenderStats.eventsNotQueuedConflatedId = type.nameToId(
+ GatewaySenderStats.EVENTS_NOT_QUEUED_CONFLATED);
+ GatewaySenderStats.eventQueueTimeId = type.nameToId(GatewaySenderStats.EVENT_QUEUE_TIME);
+ GatewaySenderStats.eventQueueSizeId = type.nameToId(GatewaySenderStats.EVENT_QUEUE_SIZE);
+ GatewaySenderStats.secondaryEventQueueSizeId = type.nameToId(
+ GatewaySenderStats.SECONDARY_EVENT_QUEUE_SIZE);
+ GatewaySenderStats.eventsProcessedByPQRMId = type.nameToId(
+ GatewaySenderStats.EVENTS_PROCESSED_BY_PQRM);
+ GatewaySenderStats.eventTmpQueueSizeId = type.nameToId(GatewaySenderStats.TMP_EVENT_QUEUE_SIZE);
+ GatewaySenderStats.eventsDistributedId = type.nameToId(GatewaySenderStats.EVENTS_DISTRIBUTED);
+ GatewaySenderStats.eventsExceedingAlertThresholdId = type.nameToId(
+ GatewaySenderStats.EVENTS_EXCEEDING_ALERT_THRESHOLD);
+ GatewaySenderStats.batchDistributionTimeId = type.nameToId(
+ GatewaySenderStats.BATCH_DISTRIBUTION_TIME);
+ GatewaySenderStats.batchesDistributedId = type.nameToId(GatewaySenderStats.BATCHES_DISTRIBUTED);
+ GatewaySenderStats.batchesRedistributedId = type.nameToId(
+ GatewaySenderStats.BATCHES_REDISTRIBUTED);
+ GatewaySenderStats.unprocessedTokensAddedByPrimaryId = type.nameToId(
+ GatewaySenderStats.UNPROCESSED_TOKENS_ADDED_BY_PRIMARY);
+ GatewaySenderStats.unprocessedEventsAddedBySecondaryId = type.nameToId(
+ GatewaySenderStats.UNPROCESSED_EVENTS_ADDED_BY_SECONDARY);
+ GatewaySenderStats.unprocessedEventsRemovedByPrimaryId = type.nameToId(
+ GatewaySenderStats.UNPROCESSED_EVENTS_REMOVED_BY_PRIMARY);
+ GatewaySenderStats.unprocessedTokensRemovedBySecondaryId = type.nameToId(
+ GatewaySenderStats.UNPROCESSED_TOKENS_REMOVED_BY_SECONDARY);
+ GatewaySenderStats.unprocessedEventsRemovedByTimeoutId = type.nameToId(
+ GatewaySenderStats.UNPROCESSED_EVENTS_REMOVED_BY_TIMEOUT);
+ GatewaySenderStats.unprocessedTokensRemovedByTimeoutId = type.nameToId(
+ GatewaySenderStats.UNPROCESSED_TOKENS_REMOVED_BY_TIMEOUT);
+ GatewaySenderStats.unprocessedEventMapSizeId = type.nameToId(
+ GatewaySenderStats.UNPROCESSED_EVENT_MAP_SIZE);
+ GatewaySenderStats.unprocessedTokenMapSizeId = type.nameToId(
+ GatewaySenderStats.UNPROCESSED_TOKEN_MAP_SIZE);
+ GatewaySenderStats.conflationIndexesMapSizeId = type.nameToId(
+ GatewaySenderStats.CONFLATION_INDEXES_MAP_SIZE);
+ GatewaySenderStats.notQueuedEventsId = type.nameToId(GatewaySenderStats.NOT_QUEUED_EVENTS);
+ GatewaySenderStats.eventsDroppedDueToPrimarySenderNotRunningId =
+ type.nameToId(GatewaySenderStats.EVENTS_DROPPED_DUE_TO_PRIMARY_SENDER_NOT_RUNNING);
+ GatewaySenderStats.eventsFilteredId = type.nameToId(GatewaySenderStats.EVENTS_FILTERED);
+ GatewaySenderStats.eventsConflatedFromBatchesId = type.nameToId(
+ GatewaySenderStats.EVENTS_CONFLATED_FROM_BATCHES);
+ GatewaySenderStats.loadBalancesCompletedId = type.nameToId(
+ GatewaySenderStats.LOAD_BALANCES_COMPLETED);
+ GatewaySenderStats.loadBalancesInProgressId = type.nameToId(
+ GatewaySenderStats.LOAD_BALANCES_IN_PROGRESS);
+ GatewaySenderStats.loadBalanceTimeId = type.nameToId(GatewaySenderStats.LOAD_BALANCE_TIME);
+ GatewaySenderStats.synchronizationEventsEnqueuedId = type.nameToId(
+ GatewaySenderStats.SYNCHRONIZATION_EVENTS_ENQUEUED);
+ GatewaySenderStats.synchronizationEventsProvidedId = type.nameToId(
+ GatewaySenderStats.SYNCHRONIZATION_EVENTS_PROVIDED);
+ }
+
+ /**
+ * Constructor.
+ * @param factory The <code>StatisticsFactory</code> which creates the <code>Statistics</code>
+ * instance
+ * @param asyncQueueId The id of the <code>AsyncEventQueue</code> used to generate the name of
+ * the
+ * <code>Statistics</code>
+ */
+ public AsyncEventQueueStats(StatisticsFactory factory, String asyncQueueId) {
+ super();
+ initializeStats(factory);
+ this.stats = factory.createAtomicStatistics(type, "asyncEventQueueStats-" + asyncQueueId);
+ }
+
+ public StatisticsType getType() {
+ return type;
+ }
+}
diff --git a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/GatewayReceiverStats.kt b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/GatewayReceiverStats.kt
new file mode 100644
index 0000000..28aadca
--- /dev/null
+++ b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/GatewayReceiverStats.kt
@@ -0,0 +1,131 @@
+/*
+ * 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.statistics.wan
+
+import org.apache.geode.statistics.cache.CacheServerStats
+import org.apache.geode.statistics.internal.micrometer.CounterStatisticMeter
+import org.apache.geode.statistics.util.NOW_NANOS
+
+//By inheritence this is a MicrometerMeterGroup
+class GatewayReceiverStats(private val owner: String) : CacheServerStats(owner,"GatewayReceiverStats-$owner") {
+
+ override fun getCommonTags(): Array<String> = arrayOf("owner", owner)
+
+ override fun initializeStaticMeters() {
+ registerMeter(gatewayReceiverBatchesDuplicateMeter)
+ registerMeter(gatewayReceiverBatchesOutOfOrderMeter)
+ registerMeter(gatewayReceiverEarlyAcksMeter)
+ registerMeter(gatewayReceiverTotalEventsReceivedMeter)
+ registerMeter(gatewayReceiverCreateEventsReceivedMeter)
+ registerMeter(gatewayReceiverUpdateEventsReceivedMeter)
+ registerMeter(gatewayReceiverDestroyEventsReceivedMeter)
+ registerMeter(gatewayReceiverUnknownEventsReceivedMeter)
+ registerMeter(gatewayReceiverExceptionsMeter)
+ registerMeter(gatewayReceiverEventsRetriedMeter)
+ }
+
+ private val gatewayReceiverBatchesDuplicateMeter = CounterStatisticMeter("gateway.receiver.batches.duplicate.count", "number of batches which have already been seen by this GatewayReceiver")
+ private val gatewayReceiverBatchesOutOfOrderMeter = CounterStatisticMeter("gateway.receiver.batches.outoforder.count", "number of batches which are out of order on this GatewayReceiver")
+ private val gatewayReceiverEarlyAcksMeter = CounterStatisticMeter("gateway.receiver.earlyAcks.count", "number of early acknowledgements sent to gatewaySenders")
+ private val gatewayReceiverTotalEventsReceivedMeter = CounterStatisticMeter("gateway.receiver.events.received.count", "total number events across the batched received by this GatewayReceiver")
+ private val gatewayReceiverCreateEventsReceivedMeter = CounterStatisticMeter("gateway.receiver.events.received.count", "total number of create operations received by this GatewayReceiver", arrayOf("operationType", "create"))
+ private val gatewayReceiverUpdateEventsReceivedMeter = CounterStatisticMeter("gateway.receiver.events.received.count", "total number of update operations received by this GatewayReceiver", arrayOf("operationType", "update"))
+ private val gatewayReceiverDestroyEventsReceivedMeter = CounterStatisticMeter("gateway.receiver.events.received.count", "total number of destroy operations received by this GatewayReceiver", arrayOf("operationType", "destroy"))
+ private val gatewayReceiverUnknownEventsReceivedMeter = CounterStatisticMeter("gateway.receiver.events.received.count", "total number of unknown operations received by this GatewayReceiver", arrayOf("operationType", "unknown"))
+ private val gatewayReceiverExceptionsMeter = CounterStatisticMeter("gateway.receiver.exceptions.count", "number of exceptions occurred while porcessing the batches")
+ private val gatewayReceiverEventsRetriedMeter = CounterStatisticMeter("gateway.receiver.events.retried", "total number events retried by this GatewayReceiver due to exceptions")
+
+ val duplicateBatchesReceived: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverBatchesDuplicateMeter.getValue()
+
+ val outoforderBatchesReceived: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverBatchesOutOfOrderMeter.getValue()
+
+ val earlyAcks: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverEarlyAcksMeter.getValue()
+
+ val eventsReceived: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverTotalEventsReceivedMeter.getValue()
+
+ val createRequest: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverCreateEventsReceivedMeter.getValue()
+
+ val updateRequest: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverUpdateEventsReceivedMeter.getValue()
+
+ val destroyRequest: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverDestroyEventsReceivedMeter.getValue()
+
+ val unknowsOperationsReceived: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverUnknownEventsReceivedMeter.getValue()
+
+ val exceptionsOccurred: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverExceptionsMeter.getValue()
+
+ val eventsRetried: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewayReceiverEventsRetriedMeter.getValue()
+
+ fun incDuplicateBatchesReceived() {
+ gatewayReceiverBatchesDuplicateMeter.increment()
+ }
+
+ fun incOutoforderBatchesReceived() {
+ gatewayReceiverBatchesOutOfOrderMeter.increment()
+ }
+
+ fun incEarlyAcks() {
+ gatewayReceiverEarlyAcksMeter.increment()
+ }
+
+ fun incEventsReceived(delta: Long) {
+ gatewayReceiverTotalEventsReceivedMeter.increment()
+ }
+
+ fun incCreateRequest() {
+ gatewayReceiverCreateEventsReceivedMeter.increment()
+ }
+
+ fun incUpdateRequest() {
+ gatewayReceiverUpdateEventsReceivedMeter.increment()
+ }
+
+ fun incDestroyRequest() {
+ gatewayReceiverDestroyEventsReceivedMeter.increment()
+ }
+
+ fun incUnknowsOperationsReceived() {
+ gatewayReceiverUnknownEventsReceivedMeter.increment()
+ }
+
+ fun incExceptionsOccurred() {
+ gatewayReceiverExceptionsMeter.increment()
+ }
+
+ fun incEventsRetried() {
+ gatewayReceiverEventsRetriedMeter.increment()
+ }
+
+ fun startTime(): Long = NOW_NANOS
+}
diff --git a/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/GatewaySenderStats.kt b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/GatewaySenderStats.kt
new file mode 100644
index 0000000..95c2907
--- /dev/null
+++ b/geode-micrometer-stats/src/main/kotlin/org/apache/geode/statistics/wan/GatewaySenderStats.kt
@@ -0,0 +1,374 @@
+/*
+ * 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.statistics.wan
+
+import org.apache.geode.statistics.internal.micrometer.CounterStatisticMeter
+import org.apache.geode.statistics.internal.micrometer.GaugeStatisticMeter
+import org.apache.geode.statistics.internal.micrometer.MicrometerMeterGroup
+import org.apache.geode.statistics.internal.micrometer.TimerStatisticMeter
+import org.apache.geode.statistics.util.NOW_NANOS
+
+class GatewaySenderStats(private val gateReceiverName: String) : MicrometerMeterGroup("GatewayReceiverStats-$gateReceiverName") {
+
+ override fun getCommonTags(): Array<String> = arrayOf("gatewaySenderName", gateReceiverName)
+
+ private val gatewaySenderEventsReceivedMeter = CounterStatisticMeter("gateway.sender.event.received.count", "Number of events received by this Sender.")
+ private val gatewaySenderEventsQueueMeter = CounterStatisticMeter("gateway.sender.event.queued.count", "Number of events added to the event queue.")
+ private val gatewaySenderEventsQueueTimer = TimerStatisticMeter("gateway.sender.event.queue.time", "Total time spent queueing events.", unit = "nanoseconds")
+ private val gatewaySenderEventQueueSizeMeter = GaugeStatisticMeter("gateway.sender.event.queue.size", "Size of the event queue.", arrayOf("queueType", "primary"))
+ private val gatewaySenderSecondaryEventQueueSizeMeter = GaugeStatisticMeter("gateway.sender.event.queue.size", "Size of secondary event queue.", arrayOf("queueType", "secondary"))
+ private val gatewaySenderEventsProcessedByPQRMMeter = GaugeStatisticMeter("gateway.sender.event.processed.pqrm.count", "Total number of events processed by Parallel Queue Removal Message(PQRM).")
+ private val gatewaySenderTempEventQueueSizeMeter = GaugeStatisticMeter("gateway.sender.event.queue.size", "Size of the temporary events.", arrayOf("queueType", "temp"))
+ private val gatewaySenderEventsNotQueuedConflatedMeter = CounterStatisticMeter("gateway.sender.event.notqueued.conflated.count", "Number of events received but not added to the event queue because the queue already contains an event with the event's key.")
+ private val gatewaySenderEventsConflatedMeter = CounterStatisticMeter("gateway.sender.event.conflated.count", "Number of events conflated from batches.")
+ private val gatewaySenderEventsDistributionMeter = CounterStatisticMeter("gateway.sender.event.distribution.count", "Number of events removed from the event queue and sent.")
+ private val gatewaySenderEventExceededAlertThresholdMeter = CounterStatisticMeter("gateway.sender.event.alertthreshold.count", "Number of events exceeding the alert threshold.")
+ private val gatewaySenderEventsDistributionTimer = TimerStatisticMeter("gateway.sender.event.distribution.time", "Total time spent distributing batches of events to other gateway receivers.", unit = "nanoseconds")
+ private val gatewaySenderEventsBatchDistributedMeter = CounterStatisticMeter("gateway.sender.batches.distributed.count", "Number of batches of events removed from the event queue and sent.")
+ private val gatewaySenderEventsBatchRedistributedMeter = CounterStatisticMeter("gateway.sender.batches.redistributed.count", "Number of batches of events removed from the event queue and resent.")
+ private val gatewaySenderEventsBatchResizedMeter = CounterStatisticMeter("gateway.sender.batches.resized.count", "Number of batches that were resized because they were too large")
+ private val gatewaySenderUnprocessedTokenAddedByPrimaryMeter = CounterStatisticMeter("gateway.sender.unprocessed.token.secondary.added.count", "Number of tokens added to the secondary's unprocessed token map by the primary (though a listener).", arrayOf("actorType", "primary"))
+ private val gatewaySenderUnprocessedEventAddedBySecondaryMeter = CounterStatisticMeter("gateway.sender.unprocessed.event.secondary.added.count", "Number of events added to the secondary's unprocessed event map by the secondary.", arrayOf("actorType", "secondary"))
+ private val gatewaySenderUnprocessedEventRemovedByPrimaryMeter = CounterStatisticMeter("gateway.sender.unprocessed.event.secondary.removed.count", "Number of events removed from the secondary's unprocessed event map by the primary (though a listener).", arrayOf("actorType", "primary"))
+ private val gatewaySenderUnprocessedTokenRemovedBySecondaryMeter = CounterStatisticMeter("gateway.sender.unprocessed.token.secondary.removed.count", "Number of tokens removed from the secondary's unprocessed token map by the secondary.", arrayOf("actorType", "secondary"))
+ private val gatewaySenderUnprocessedEventRemovedByTimeoutMeter = CounterStatisticMeter("gateway.sender.unprocessed.event.secondary.removed.count", "Number of events removed from the secondary's unprocessed event map by a timeout.", arrayOf("actorType", "timeout"))
+ private val gatewaySenderUnprocessedTokenRemovedByTimeoutMeter = CounterStatisticMeter("gateway.sender.unprocessed.token.secondary.removed.count", "Number of tokens removed from the secondary's unprocessed token map by a timeout.", arrayOf("actorType", "timeout"))
+ private val gatewaySenderUnprocessedEventSizeMeter = GaugeStatisticMeter("gateway.sender.unprocessed.event.secondary.size", "Current number of entries in the secondary's unprocessed event map.")
+ private val gatewaySenderUnprocessedTokenSizeMeter = GaugeStatisticMeter("gateway.sender.unprocessed.token.secondary.size", "Current number of entries in the secondary's unprocessed token map.")
+ private val gatewaySenderConflationIndexSizeMeter = GaugeStatisticMeter("gateway.sender.conflation.index.size", "Current number of entries in the conflation indexes map.")
+ private val gatewaySenderEventsNotQueuedMeter = CounterStatisticMeter("gateway.sender.notQueued.count", "Number of events not added to queue.")
+ private val gatewaySenderEventsDroppedSenderNotRunningMeter = CounterStatisticMeter("gateway.sender.events.dropped.count", "Number of events dropped because the primary gateway sender is not running.", arrayOf("reason", "primarySenderNotRunning"))
+ private val gatewaySenderEventsDroppedFilteredMeter = CounterStatisticMeter("gateway.sender.events.filtered.count", "Number of events filtered through GatewayEventFilter.")
+ private val gatewaySenderLoadBalancesCompletedMeter = CounterStatisticMeter("gateway.sender.loadBalances.completed.count", "Number of load balances completed")
+ private val gatewaySenderLoadBalancesInProgressMeter = GaugeStatisticMeter("gateway.sender.loadBalances.inprogress.count", "Number of load balances in progress")
+ private val gatewaySenderLoadBalancesTimer = TimerStatisticMeter("gateway.sender.loadBalances.time", "Total time spent load balancing this sender", unit = "nanoseconds")
+ private val gatewaySenderSynchronizationEventsQueuedMeter = CounterStatisticMeter("gateway.sender.synchronization.events.queued.count", "Number of synchronization events added to the event queue.")
+ private val gatewaySenderSynchronizationEventsSentMeter = CounterStatisticMeter("gateway.sender.synchronization.events.sent.count", "Number of synchronization events provided to other members.")
+
+ override fun initializeStaticMeters() {
+ registerMeter(gatewaySenderEventsReceivedMeter)
+ registerMeter(gatewaySenderEventsQueueMeter)
+ registerMeter(gatewaySenderEventsQueueTimer)
+ registerMeter(gatewaySenderEventQueueSizeMeter)
+ registerMeter(gatewaySenderSecondaryEventQueueSizeMeter)
+ registerMeter(gatewaySenderEventsProcessedByPQRMMeter)
+ registerMeter(gatewaySenderTempEventQueueSizeMeter)
+ registerMeter(gatewaySenderEventsNotQueuedConflatedMeter)
+ registerMeter(gatewaySenderEventsConflatedMeter)
+ registerMeter(gatewaySenderEventsDistributionMeter)
+ registerMeter(gatewaySenderEventExceededAlertThresholdMeter)
+ registerMeter(gatewaySenderEventsDistributionTimer)
+ registerMeter(gatewaySenderEventsBatchDistributedMeter)
+ registerMeter(gatewaySenderEventsBatchRedistributedMeter)
+ registerMeter(gatewaySenderEventsBatchResizedMeter)
+ registerMeter(gatewaySenderUnprocessedTokenAddedByPrimaryMeter)
+ registerMeter(gatewaySenderUnprocessedEventAddedBySecondaryMeter)
+ registerMeter(gatewaySenderUnprocessedEventRemovedByPrimaryMeter)
+ registerMeter(gatewaySenderUnprocessedTokenRemovedBySecondaryMeter)
+ registerMeter(gatewaySenderUnprocessedEventRemovedByTimeoutMeter)
+ registerMeter(gatewaySenderUnprocessedTokenRemovedByTimeoutMeter)
+ registerMeter(gatewaySenderUnprocessedEventSizeMeter)
+ registerMeter(gatewaySenderUnprocessedTokenSizeMeter)
+ registerMeter(gatewaySenderConflationIndexSizeMeter)
+ registerMeter(gatewaySenderEventsNotQueuedMeter)
+ registerMeter(gatewaySenderEventsDroppedSenderNotRunningMeter)
+ registerMeter(gatewaySenderEventsDroppedFilteredMeter)
+ registerMeter(gatewaySenderLoadBalancesCompletedMeter)
+ registerMeter(gatewaySenderLoadBalancesInProgressMeter)
+ registerMeter(gatewaySenderLoadBalancesTimer)
+ registerMeter(gatewaySenderSynchronizationEventsQueuedMeter)
+ registerMeter(gatewaySenderSynchronizationEventsSentMeter)
+ }
+
+ val eventsReceived: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsReceivedMeter.getValue()
+
+ val eventsQueued: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsQueueMeter.getValue()
+
+ val eventsNotQueuedConflated: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsNotQueuedConflatedMeter.getValue()
+
+ val eventsConflatedFromBatches: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsConflatedMeter.getValue()
+
+ val eventQueueSize: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventQueueSizeMeter.getValue()
+
+ val secondaryEventQueueSize: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderSecondaryEventQueueSizeMeter.getValue()
+
+ var eventsProcessedByPQRM: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsProcessedByPQRMMeter.getValue()
+ set(size) {
+ gatewaySenderEventsProcessedByPQRMMeter.setValue(size)
+ }
+
+ val tempEventQueueSize: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderTempEventQueueSizeMeter.getValue()
+
+
+ val eventsDistributed: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsDistributionMeter.getValue()
+
+
+ val eventsExceedingAlertThreshold: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventExceededAlertThresholdMeter.getValue()
+
+ val batchesDistributed: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsBatchDistributedMeter.getValue()
+
+ val batchesRedistributed: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsBatchRedistributedMeter.getValue()
+
+ val batchesResized: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsBatchResizedMeter.getValue()
+
+ val unprocessedTokensAddedByPrimary: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderUnprocessedTokenAddedByPrimaryMeter.getValue()
+
+ val unprocessedEventsAddedBySecondary: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderUnprocessedEventAddedBySecondaryMeter.getValue()
+
+ val unprocessedEventsRemovedByPrimary: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderUnprocessedEventRemovedByPrimaryMeter.getValue()
+
+ val unprocessedTokensRemovedBySecondary: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderUnprocessedTokenRemovedBySecondaryMeter.getValue()
+
+ val unprocessedEventMapSize: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderUnprocessedEventSizeMeter.getValue()
+
+ val unprocessedTokenMapSize: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderUnprocessedTokenSizeMeter.getValue()
+
+ val eventsNotQueued: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsNotQueuedMeter.getValue()
+
+ val eventsDroppedDueToPrimarySenderNotRunning: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsDroppedSenderNotRunningMeter.getValue()
+
+ val eventsFiltered: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderEventsDroppedFilteredMeter.getValue()
+
+ val conflationIndexesMapSize: Long
+ @Deprecated(message = "This was just done as an interim solution until GEODE does not depend on stats for internal state")
+ get() = gatewaySenderConflationIndexSizeMeter.getValue()
+
+ fun incEventsReceived() {
+ gatewaySenderEventsReceivedMeter.increment()
+ }
+
+ fun incEventsExceedingAlertThreshold() {
+ gatewaySenderEventExceededAlertThresholdMeter.increment()
+ }
+
+ fun incBatchesRedistributed() {
+ gatewaySenderEventsBatchDistributedMeter.increment()
+ }
+
+ fun incBatchesResized() {
+ gatewaySenderEventsBatchResizedMeter.increment()
+ }
+
+ fun setQueueSize(size: Long) {
+ gatewaySenderEventQueueSizeMeter.setValue(size)
+ }
+
+ fun setSecondaryQueueSize(size: Long) {
+ gatewaySenderSecondaryEventQueueSizeMeter.setValue(size)
+ }
+
+ fun setTempQueueSize(size: Long) {
+ gatewaySenderTempEventQueueSizeMeter.setValue(size)
+ }
+
+ @JvmOverloads
+ fun incQueueSize(delta: Long = 1) {
+ gatewaySenderEventQueueSizeMeter.increment(delta)
+ }
+
+ @JvmOverloads
+ fun incSecondaryQueueSize(delta: Long = 1) {
+ gatewaySenderSecondaryEventQueueSizeMeter.increment(delta)
+ }
+
+ @JvmOverloads
+ fun incTempQueueSize(delta: Long = 1) {
+ gatewaySenderTempEventQueueSizeMeter.increment(delta)
+ }
+
+ fun incEventsProcessedByPQRM(delta: Long) {
+ gatewaySenderEventsProcessedByPQRMMeter.increment(delta)
+ }
+
+
+ @JvmOverloads
+ fun decQueueSize(delta: Long = 1) {
+ gatewaySenderEventQueueSizeMeter.increment(delta)
+ }
+
+ @JvmOverloads
+ fun decSecondaryQueueSize(delta: Long = 1) {
+ gatewaySenderSecondaryEventQueueSizeMeter.increment(delta)
+ }
+
+ @JvmOverloads
+ fun decTempQueueSize(delta: Long = 1) {
+ gatewaySenderTempEventQueueSizeMeter.increment(delta)
+ }
+
+ fun incEventsNotQueuedConflated() {
+ gatewaySenderEventsNotQueuedConflatedMeter.increment()
+ }
+
+ fun incEventsConflatedFromBatches(numEvents: Long) {
+ gatewaySenderEventsConflatedMeter.increment(numEvents)
+ }
+
+ fun incEventsNotQueued() {
+ gatewaySenderEventsNotQueuedMeter.increment()
+ }
+
+ fun incEventsDroppedDueToPrimarySenderNotRunning() {
+ gatewaySenderEventsDroppedSenderNotRunningMeter.increment()
+ }
+
+ fun incEventsFiltered() {
+ gatewaySenderEventsDroppedFilteredMeter.increment()
+ }
+
+ fun incUnprocessedTokensAddedByPrimary() {
+ gatewaySenderUnprocessedTokenAddedByPrimaryMeter.increment()
+ incUnprocessedTokenMapSize()
+ }
+
+ fun incUnprocessedEventsAddedBySecondary() {
+ gatewaySenderUnprocessedEventAddedBySecondaryMeter.increment()
+ incUnprocessedEventMapSize()
+ }
+
+ fun incUnprocessedEventsRemovedByPrimary() {
+ gatewaySenderUnprocessedEventRemovedByPrimaryMeter.increment()
+ decUnprocessedEventMapSize()
+ }
+
+ fun incUnprocessedTokensRemovedBySecondary() {
+ gatewaySenderUnprocessedTokenRemovedBySecondaryMeter.increment()
+ decUnprocessedTokenMapSize()
+ }
+
+ fun incUnprocessedEventsRemovedByTimeout(count: Long) {
+ gatewaySenderUnprocessedEventRemovedByTimeoutMeter.increment(count)
+ decUnprocessedEventMapSize(count)
+ }
+
+ fun incUnprocessedTokensRemovedByTimeout(count: Long) {
+ gatewaySenderUnprocessedTokenRemovedByTimeoutMeter.increment(count)
+ decUnprocessedTokenMapSize(count)
+ }
+
+ fun clearUnprocessedMaps() {
+ gatewaySenderUnprocessedEventSizeMeter.setValue(0)
+ gatewaySenderUnprocessedTokenSizeMeter.setValue(0)
+ }
+
+ private fun incUnprocessedEventMapSize() {
+ gatewaySenderUnprocessedEventSizeMeter.increment()
+ }
+
+ @JvmOverloads
+ private fun decUnprocessedEventMapSize(count: Long = 1) {
+ gatewaySenderUnprocessedEventSizeMeter.decrement(count)
+ }
+
+
+ private fun incUnprocessedTokenMapSize() {
+ gatewaySenderUnprocessedTokenSizeMeter
+ }
+
+ @JvmOverloads
+ private fun decUnprocessedTokenMapSize(decCount: Long = 1) {
+ gatewaySenderUnprocessedTokenSizeMeter.decrement(decCount)
+ }
+
+ fun incConflationIndexesMapSize() {
+ gatewaySenderConflationIndexSizeMeter.increment()
+ }
+
+ fun decConflationIndexesMapSize() {
+ gatewaySenderConflationIndexSizeMeter.decrement()
+ }
+
+ fun startTime(): Long = NOW_NANOS
+
+ fun endBatch(start: Long, numberOfEvents: Long) {
+ // Increment event distribution time
+ gatewaySenderEventsDistributionTimer.recordValue(NOW_NANOS - start)
+
+ // Increment number of batches distributed
+ gatewaySenderEventsBatchDistributedMeter.increment()
+
+ // Increment number of events distributed
+ gatewaySenderEventsDistributionMeter.increment()
+ }
+
+ fun endPut(start: Long) {
+ gatewaySenderEventsQueueTimer.recordValue(NOW_NANOS - start)
+ // Increment number of event queued
+ gatewaySenderEventsQueueMeter.increment()
+ }
+
+ fun startLoadBalance(): Long {
+ gatewaySenderLoadBalancesInProgressMeter.increment()
+ return NOW_NANOS
+ }
+
+ fun endLoadBalance(start: Long) {
+ gatewaySenderLoadBalancesTimer.recordValue(NOW_NANOS - start)
+ gatewaySenderLoadBalancesInProgressMeter.decrement()
+ gatewaySenderLoadBalancesCompletedMeter.increment()
+ }
+
+ fun incSynchronizationEventsEnqueued() {
+ gatewaySenderSynchronizationEventsQueuedMeter.increment()
+ }
+
+ fun incSynchronizationEventsProvided() {
+ gatewaySenderSynchronizationEventsSentMeter.increment()
+ }
+}