You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tomcat.apache.org by ma...@apache.org on 2024/05/10 12:33:40 UTC
(tomcat) branch main updated: Code clean-up - no functional change
This is an automated email from the ASF dual-hosted git repository.
markt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/main by this push:
new b595fdfad6 Code clean-up - no functional change
b595fdfad6 is described below
commit b595fdfad68a22e682f16d002d958439e3f42dc9
Author: Mark Thomas <ma...@apache.org>
AuthorDate: Fri May 10 13:33:30 2024 +0100
Code clean-up - no functional change
---
.../interceptors/DomainFilterInterceptor.java | 33 +-
.../group/interceptors/EncryptInterceptor.java | 178 +++----
.../interceptors/EncryptInterceptorMBean.java | 6 +
.../interceptors/FragmentationInterceptor.java | 112 ++---
.../tribes/group/interceptors/GzipInterceptor.java | 27 +-
.../group/interceptors/GzipInterceptorMBean.java | 36 +-
.../interceptors/MessageDispatchInterceptor.java | 66 ++-
.../group/interceptors/NonBlockingCoordinator.java | 521 +++++++++++----------
.../group/interceptors/OrderInterceptor.java | 149 +++---
.../group/interceptors/SimpleCoordinator.java | 6 +-
.../interceptors/StaticMembershipInterceptor.java | 44 +-
.../StaticMembershipInterceptorMBean.java | 2 +-
.../group/interceptors/TcpFailureDetector.java | 161 ++++---
.../group/interceptors/TcpPingInterceptor.java | 74 ++-
.../group/interceptors/ThroughputInterceptor.java | 38 +-
.../interceptors/TwoPhaseCommitInterceptor.java | 74 +--
16 files changed, 772 insertions(+), 755 deletions(-)
diff --git a/java/org/apache/catalina/tribes/group/interceptors/DomainFilterInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/DomainFilterInterceptor.java
index c0c7b06124..4d4fbf0918 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/DomainFilterInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/DomainFilterInterceptor.java
@@ -30,8 +30,7 @@ import org.apache.juli.logging.LogFactory;
/**
* Filters membership based on domain.
*/
-public class DomainFilterInterceptor extends ChannelInterceptorBase
- implements DomainFilterInterceptorMBean {
+public class DomainFilterInterceptor extends ChannelInterceptorBase implements DomainFilterInterceptorMBean {
private static final Log log = LogFactory.getLog(DomainFilterInterceptor.class);
protected static final StringManager sm = StringManager.getManager(DomainFilterInterceptor.class);
@@ -53,25 +52,25 @@ public class DomainFilterInterceptor extends ChannelInterceptorBase
}
}
}
- }//messageReceived
+ }// messageReceived
@Override
public void memberAdded(Member member) {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
boolean notify = false;
synchronized (membership) {
- notify = Arrays.equals(domain,member.getDomain());
- if ( notify ) {
+ notify = Arrays.equals(domain, member.getDomain());
+ if (notify) {
notify = membership.memberAlive(member);
}
}
- if ( notify ) {
+ if (notify) {
super.memberAdded(member);
} else {
- if(log.isInfoEnabled()) {
+ if (log.isInfoEnabled()) {
log.info(sm.getString("domainFilterInterceptor.member.refused", member));
}
}
@@ -79,24 +78,24 @@ public class DomainFilterInterceptor extends ChannelInterceptorBase
@Override
public void memberDisappeared(Member member) {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
boolean notify = false;
synchronized (membership) {
- notify = Arrays.equals(domain,member.getDomain());
- if ( notify ) {
+ notify = Arrays.equals(domain, member.getDomain());
+ if (notify) {
membership.removeMember(member);
}
}
- if ( notify ) {
+ if (notify) {
super.memberDisappeared(member);
}
}
@Override
public boolean hasMembers() {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
return membership.hasMembers();
@@ -104,7 +103,7 @@ public class DomainFilterInterceptor extends ChannelInterceptorBase
@Override
public Member[] getMembers() {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
return membership.getMembers();
@@ -112,7 +111,7 @@ public class DomainFilterInterceptor extends ChannelInterceptorBase
@Override
public Member getMember(Member mbr) {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
return membership.getMember(mbr);
@@ -125,7 +124,7 @@ public class DomainFilterInterceptor extends ChannelInterceptorBase
protected synchronized void setupMembership() {
- if ( membership == null ) {
+ if (membership == null) {
membership = new Membership(super.getLocalMember(true));
}
@@ -141,7 +140,7 @@ public class DomainFilterInterceptor extends ChannelInterceptorBase
}
public void setDomain(String domain) {
- if ( domain == null ) {
+ if (domain == null) {
return;
}
if (domain.startsWith("{")) {
diff --git a/java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptor.java
index a209864e45..8c3a5cf3bf 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptor.java
@@ -43,15 +43,10 @@ import org.apache.juli.logging.LogFactory;
/**
- * Adds encryption using a pre-shared key.
- *
- * The length of the key (in bytes) must be acceptable for the encryption
- * algorithm being used. For example, for AES, you must use a key of either
- * 16 bytes (128 bits, 24 bytes 192 bits), or 32 bytes (256 bits).
- *
- * You can supply the raw key bytes by calling {@link #setEncryptionKey(byte[])}
- * or the hex-encoded binary bytes by calling
- * {@link #setEncryptionKey(String)}.
+ * Adds encryption using a pre-shared key. The length of the key (in bytes) must be acceptable for the encryption
+ * algorithm being used. For example, for AES, you must use a key of either 16 bytes (128 bits, 24 bytes 192 bits), or
+ * 32 bytes (256 bits). You can supply the raw key bytes by calling {@link #setEncryptionKey(byte[])} or the hex-encoded
+ * binary bytes by calling {@link #setEncryptionKey(String)}.
*/
public class EncryptInterceptor extends ChannelInterceptorBase implements EncryptInterceptorMBean {
@@ -75,10 +70,9 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
public void start(int svc) throws ChannelException {
validateChannelChain();
- if(Channel.SND_TX_SEQ == (svc & Channel.SND_TX_SEQ)) {
+ if (Channel.SND_TX_SEQ == (svc & Channel.SND_TX_SEQ)) {
try {
- encryptionManager = createEncryptionManager(getEncryptionAlgorithm(),
- getEncryptionKeyInternal(),
+ encryptionManager = createEncryptionManager(getEncryptionAlgorithm(), getEncryptionKeyInternal(),
getProviderName());
} catch (GeneralSecurityException gse) {
throw new ChannelException(sm.getString("encryptInterceptor.init.failed"), gse);
@@ -90,8 +84,8 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
private void validateChannelChain() throws ChannelException {
ChannelInterceptor interceptor = getPrevious();
- while(null != interceptor) {
- if(interceptor instanceof TcpFailureDetector) {
+ while (null != interceptor) {
+ if (interceptor instanceof TcpFailureDetector) {
throw new ChannelConfigException(sm.getString("encryptInterceptor.tcpFailureDetector.ordering"));
}
@@ -101,7 +95,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
@Override
public void stop(int svc) throws ChannelException {
- if(Channel.SND_TX_SEQ == (svc & Channel.SND_TX_SEQ)) {
+ if (Channel.SND_TX_SEQ == (svc & Channel.SND_TX_SEQ)) {
encryptionManager.shutdown();
}
@@ -152,28 +146,25 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
/**
- * Sets the encryption algorithm to be used for encrypting and decrypting
- * channel messages. You must specify the <code>algorithm/mode/padding</code>.
- * Information on standard algorithm names may be found in the
+ * Sets the encryption algorithm to be used for encrypting and decrypting channel messages. You must specify the
+ * <code>algorithm/mode/padding</code>. Information on standard algorithm names may be found in the
* <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/StandardNames.html">Java
- * documentation</a>.
- *
- * Default is <code>AES/CBC/PKCS5Padding</code>.
+ * documentation</a>. Default is <code>AES/CBC/PKCS5Padding</code>.
*
* @param algorithm The algorithm to use.
*/
@Override
public void setEncryptionAlgorithm(String algorithm) {
- if(null == getEncryptionAlgorithm()) {
+ if (null == getEncryptionAlgorithm()) {
throw new IllegalStateException(sm.getString("encryptInterceptor.algorithm.required"));
}
int pos = algorithm.indexOf('/');
- if(pos < 0) {
+ if (pos < 0) {
throw new IllegalArgumentException(sm.getString("encryptInterceptor.algorithm.required"));
}
pos = algorithm.indexOf('/', pos + 1);
- if(pos < 0) {
+ if (pos < 0) {
throw new IllegalArgumentException(sm.getString("encryptInterceptor.algorithm.required"));
}
@@ -181,8 +172,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
/**
- * Gets the encryption algorithm being used to encrypt and decrypt channel
- * messages.
+ * Gets the encryption algorithm being used to encrypt and decrypt channel messages.
*
* @return The algorithm being used, including the algorithm mode and padding.
*/
@@ -192,8 +182,8 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
/**
- * Sets the encryption key for encryption and decryption. The length of the
- * key must be appropriate for the algorithm being used.
+ * Sets the encryption key for encryption and decryption. The length of the key must be appropriate for the
+ * algorithm being used.
*
* @param key The encryption key.
*/
@@ -207,17 +197,16 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
/**
- * Gets the encryption key being used for encryption and decryption.
- * The key is encoded using hex-encoding where e.g. the byte <code>0xab</code>
- * will be shown as "ab". The length of the string in characters will
- * be twice the length of the key in bytes.
+ * Gets the encryption key being used for encryption and decryption. The key is encoded using hex-encoding where
+ * e.g. the byte <code>0xab</code> will be shown as "ab". The length of the string in characters will be twice the
+ * length of the key in bytes.
*
* @param keyBytes The encryption key.
*/
public void setEncryptionKey(String keyBytes) {
this.encryptionKeyString = keyBytes;
if (null == keyBytes) {
- setEncryptionKey((byte[])null);
+ setEncryptionKey((byte[]) null);
} else {
setEncryptionKey(fromHexString(keyBytes.trim()));
}
@@ -232,7 +221,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
public byte[] getEncryptionKey() {
byte[] key = getEncryptionKeyInternal();
- if(null != key) {
+ if (null != key) {
key = key.clone();
}
@@ -252,9 +241,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
/**
- * Sets the JCA provider name used for cryptographic activities.
- *
- * Default is the JVM platform default.
+ * Sets the JCA provider name used for cryptographic activities. Default is the JVM platform default.
*
* @param provider The name of the JCA provider.
*/
@@ -264,9 +251,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
/**
- * Gets the JCA provider name used for cryptographic activities.
- *
- * Default is the JVM platform default.
+ * Gets the JCA provider name used for cryptographic activities. Default is the JVM platform default.
*
* @return The name of the JCA provider.
*/
@@ -276,13 +261,14 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
// Copied from org.apache.tomcat.util.buf.HexUtils
-
+ // @formatter:off
private static final int[] DEC = {
00, 01, 02, 03, 04, 05, 06, 07, 8, 9, -1, -1, -1, -1, -1, -1,
-1, 10, 11, 12, 13, 14, 15, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-1, 10, 11, 12, 13, 14, 15,
};
+ // @formatter:on
private static int getDec(int index) {
@@ -308,8 +294,8 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
char[] inputChars = input.toCharArray();
byte[] result = new byte[input.length() >> 1];
for (int i = 0; i < result.length; i++) {
- int upperNibble = getDec(inputChars[2*i]);
- int lowerNibble = getDec(inputChars[2*i + 1]);
+ int upperNibble = getDec(inputChars[2 * i]);
+ int lowerNibble = getDec(inputChars[2 * i + 1]);
if (upperNibble < 0 || lowerNibble < 0) {
// Non hex character
throw new IllegalArgumentException(sm.getString("hexUtils.fromHex.nonHex"));
@@ -319,10 +305,9 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
return result;
}
- private static BaseEncryptionManager createEncryptionManager(String algorithm,
- byte[] encryptionKey, String providerName)
- throws NoSuchAlgorithmException, NoSuchPaddingException, NoSuchProviderException {
- if(null == encryptionKey) {
+ private static BaseEncryptionManager createEncryptionManager(String algorithm, byte[] encryptionKey,
+ String providerName) throws NoSuchAlgorithmException, NoSuchPaddingException, NoSuchProviderException {
+ if (null == encryptionKey) {
throw new IllegalStateException(sm.getString("encryptInterceptor.key.required"));
}
@@ -333,30 +318,28 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
// take just the algorithm part.
int pos = algorithm.indexOf('/');
- if(pos >= 0) {
+ if (pos >= 0) {
algorithmName = algorithm.substring(0, pos);
- int pos2 = algorithm.indexOf('/', pos+1);
+ int pos2 = algorithm.indexOf('/', pos + 1);
- if(pos2 >= 0) {
+ if (pos2 >= 0) {
algorithmMode = algorithm.substring(pos + 1, pos2);
} else {
algorithmMode = "CBC";
}
} else {
- algorithmName = algorithm;
+ algorithmName = algorithm;
algorithmMode = "CBC";
}
- if("GCM".equalsIgnoreCase(algorithmMode)) {
+ if ("GCM".equalsIgnoreCase(algorithmMode)) {
return new GCMEncryptionManager(algorithm, new SecretKeySpec(encryptionKey, algorithmName), providerName);
- } else if("CBC".equalsIgnoreCase(algorithmMode)
- || "OFB".equalsIgnoreCase(algorithmMode)
- || "CFB".equalsIgnoreCase(algorithmMode)) {
- return new BaseEncryptionManager(algorithm,
- new SecretKeySpec(encryptionKey, algorithmName),
- providerName);
+ } else if ("CBC".equalsIgnoreCase(algorithmMode) || "OFB".equalsIgnoreCase(algorithmMode) ||
+ "CFB".equalsIgnoreCase(algorithmMode)) {
+ return new BaseEncryptionManager(algorithm, new SecretKeySpec(encryptionKey, algorithmName), providerName);
} else {
- throw new IllegalArgumentException(sm.getString("encryptInterceptor.algorithm.unsupported-mode", algorithmMode));
+ throw new IllegalArgumentException(
+ sm.getString("encryptInterceptor.algorithm.unsupported-mode", algorithmMode));
}
}
@@ -382,20 +365,19 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
private final SecretKeySpec secretKey;
/**
- * A pool of Cipher objects. Ciphers are expensive to create, but not
- * to re-initialize, so we use a pool of them which grows as necessary.
+ * A pool of Cipher objects. Ciphers are expensive to create, but not to re-initialize, so we use a pool of them
+ * which grows as necessary.
*/
private final ConcurrentLinkedQueue<Cipher> cipherPool;
/**
- * A pool of SecureRandom objects. Each encrypt operation requires access
- * to a source of randomness. SecureRandom is thread-safe, but sharing a
- * single instance will likely be a bottleneck.
+ * A pool of SecureRandom objects. Each encrypt operation requires access to a source of randomness.
+ * SecureRandom is thread-safe, but sharing a single instance will likely be a bottleneck.
*/
private final ConcurrentLinkedQueue<SecureRandom> randomPool;
BaseEncryptionManager(String algorithm, SecretKeySpec secretKey, String providerName)
- throws NoSuchAlgorithmException, NoSuchPaddingException, NoSuchProviderException {
+ throws NoSuchAlgorithmException, NoSuchPaddingException, NoSuchProviderException {
this.algorithm = algorithm;
this.providerName = providerName;
this.secretKey = secretKey;
@@ -422,9 +404,8 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
/**
- * Gets the size, in bytes, of the initialization vector for the
- * cipher being used. The IV size is often, but not always, the block
- * size for the cipher.
+ * Gets the size, in bytes, of the initialization vector for the cipher being used. The IV size is often, but
+ * not always, the block size for the cipher.
*
* @return The size of the initialization vector for this algorithm.
*/
@@ -436,11 +417,10 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
return providerName;
}
- private Cipher createCipher()
- throws NoSuchAlgorithmException, NoSuchPaddingException, NoSuchProviderException {
+ private Cipher createCipher() throws NoSuchAlgorithmException, NoSuchPaddingException, NoSuchProviderException {
String providerName = getProviderName();
- if(null == providerName) {
+ if (null == providerName) {
return Cipher.getInstance(getAlgorithm());
} else {
return Cipher.getInstance(getAlgorithm(), providerName);
@@ -450,7 +430,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
private Cipher getCipher() throws GeneralSecurityException {
Cipher cipher = cipherPool.poll();
- if(null == cipher) {
+ if (null == cipher) {
cipher = createCipher();
}
@@ -464,7 +444,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
private SecureRandom getRandom() {
SecureRandom random = randomPool.poll();
- if(null == random) {
+ if (null == random) {
random = new SecureRandom();
}
@@ -476,13 +456,10 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
/**
- * Encrypts the input <code>bytes</code> into two separate byte arrays:
- * one for the random initialization vector (IV) used for this message,
- * and the second one containing the actual encrypted payload.
- *
- * This method returns a pair of byte arrays instead of a single
- * concatenated one to reduce the number of byte buffers created
- * and copied during the whole operation -- including message re-building.
+ * Encrypts the input <code>bytes</code> into two separate byte arrays: one for the random initialization vector
+ * (IV) used for this message, and the second one containing the actual encrypted payload. This method returns a
+ * pair of byte arrays instead of a single concatenated one to reduce the number of byte buffers created and
+ * copied during the whole operation -- including message re-building.
*
* @param bytes The data to encrypt.
*
@@ -509,7 +486,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
return data;
} finally {
- if(null != cipher) {
+ if (null != cipher) {
returnCipher(cipher);
}
}
@@ -538,7 +515,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
// Decrypt remainder of the message.
return cipher.doFinal(bytes, ivSize, bytes.length - ivSize);
} finally {
- if(null != cipher) {
+ if (null != cipher) {
returnCipher(cipher);
}
}
@@ -559,7 +536,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
return ivBytes;
} finally {
- if(null != random) {
+ if (null != random) {
returnRandom(random);
}
}
@@ -572,25 +549,20 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
/**
* Implements an EncryptionManager for using GCM block cipher modes.
- *
- * GCM works a little differently than some of the other block cipher modes
- * supported by EncryptInterceptor. First of all, it requires a different
- * kind of AlgorithmParameterSpec object to be used, and second, it
- * requires a slightly different initialization vector and something called
- * an "authentication tag".
- *
- * The choice of IV length can be somewhat arbitrary, but there is consensus
- * that 96-bit (12-byte) IVs for GCM are the best trade-off between security
- * and performance. For other block cipher modes, IV length is the same as
- * the block size.
- *
- * The "authentication tag" is a computed authentication value based upon
- * the message and the encryption process. GCM defines these tags as the
- * number of bits to use for the authentication tag, and it's clear that
- * the highest number of bits supported 128-bit provide the best security.
+ * <p>
+ * GCM works a little differently than some of the other block cipher modes supported by EncryptInterceptor. First
+ * of all, it requires a different kind of AlgorithmParameterSpec object to be used, and second, it requires a
+ * slightly different initialization vector and something called an "authentication tag".
+ * <p>
+ * The choice of IV length can be somewhat arbitrary, but there is consensus that 96-bit (12-byte) IVs for GCM are
+ * the best trade-off between security and performance. For other block cipher modes, IV length is the same as the
+ * block size.
+ * <p>
+ * The "authentication tag" is a computed authentication value based upon the message and the encryption process.
+ * GCM defines these tags as the number of bits to use for the authentication tag, and it's clear that the highest
+ * number of bits supported 128-bit provide the best security.
*/
- private static class GCMEncryptionManager extends BaseEncryptionManager
- {
+ private static class GCMEncryptionManager extends BaseEncryptionManager {
GCMEncryptionManager(String algorithm, SecretKeySpec secretKey, String providerName)
throws NoSuchAlgorithmException, NoSuchPaddingException, NoSuchProviderException {
super(algorithm, secretKey, providerName);
@@ -608,9 +580,7 @@ public class EncryptInterceptor extends ChannelInterceptorBase implements Encryp
}
}
- static class ChannelConfigException
- extends ChannelException
- {
+ static class ChannelConfigException extends ChannelException {
private static final long serialVersionUID = 1L;
ChannelConfigException(String message) {
diff --git a/java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptorMBean.java b/java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptorMBean.java
index 04c3b4e6e9..b5ed1001d8 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptorMBean.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptorMBean.java
@@ -20,12 +20,18 @@ public interface EncryptInterceptorMBean {
// Config
int getOptionFlag();
+
void setOptionFlag(int optionFlag);
void setEncryptionAlgorithm(String algorithm);
+
String getEncryptionAlgorithm();
+
void setEncryptionKey(byte[] key);
+
byte[] getEncryptionKey();
+
void setProviderName(String provider);
+
String getProviderName();
}
diff --git a/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
index 98c8353f6f..559f0c59ef 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java
@@ -32,29 +32,30 @@ import org.apache.juli.logging.LogFactory;
/**
* The fragmentation interceptor splits up large messages into smaller messages and assembles them on the other end.
- * This is very useful when you don't want large messages hogging the sending sockets
- * and smaller messages can make it through.
- *
- * <br><b>Configuration Options</b><br>
- * FragmentationInterceptor.expire=<milliseconds> - how long do we keep the fragments in memory and wait for the rest to arrive <b>default=60,000ms -> 60seconds</b>
- * This setting is useful to avoid OutOfMemoryErrors<br>
- * FragmentationInterceptor.maxSize=<max message size> - message size in bytes <b>default=1024*100 (around a tenth of a MB)</b><br>
+ * This is very useful when you don't want large messages hogging the sending sockets and smaller messages can make it
+ * through. <br>
+ * <b>Configuration Options</b><br>
+ * FragmentationInterceptor.expire=<milliseconds> - how long do we keep the fragments in memory and wait for the
+ * rest to arrive <b>default=60,000ms -> 60seconds</b> This setting is useful to avoid OutOfMemoryErrors<br>
+ * FragmentationInterceptor.maxSize=<max message size> - message size in bytes <b>default=1024*100 (around a tenth
+ * of a MB)</b><br>
*/
public class FragmentationInterceptor extends ChannelInterceptorBase implements FragmentationInterceptorMBean {
private static final Log log = LogFactory.getLog(FragmentationInterceptor.class);
protected static final StringManager sm = StringManager.getManager(FragmentationInterceptor.class);
- protected final HashMap<FragKey, FragCollection> fragpieces = new HashMap<>();
- private int maxSize = 1024*100;
- private long expire = 1000 * 60; //one minute expiration
+ protected final HashMap<FragKey,FragCollection> fragpieces = new HashMap<>();
+ private int maxSize = 1024 * 100;
+ private long expire = 1000 * 60; // one minute expiration
protected final boolean deepclone = true;
@Override
- public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
+ public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
+ throws ChannelException {
int size = msg.getMessage().getLength();
- boolean frag = (size>maxSize) && okToProcess(msg.getOptions());
- if ( frag ) {
+ boolean frag = (size > maxSize) && okToProcess(msg.getOptions());
+ if (frag) {
frag(destination, msg, payload);
} else {
msg.getMessage().append(frag);
@@ -64,9 +65,9 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements
@Override
public void messageReceived(ChannelMessage msg) {
- boolean isFrag = XByteBuffer.toBoolean(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-1);
+ boolean isFrag = XByteBuffer.toBoolean(msg.getMessage().getBytesDirect(), msg.getMessage().getLength() - 1);
msg.getMessage().trim(1);
- if ( isFrag ) {
+ if (isFrag) {
defrag(msg);
} else {
super.messageReceived(msg);
@@ -76,10 +77,10 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements
public FragCollection getFragCollection(FragKey key, ChannelMessage msg) {
FragCollection coll = fragpieces.get(key);
- if ( coll == null ) {
+ if (coll == null) {
synchronized (fragpieces) {
coll = fragpieces.get(key);
- if ( coll == null ) {
+ if (coll == null) {
coll = new FragCollection(msg);
fragpieces.put(key, coll);
}
@@ -92,12 +93,12 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements
fragpieces.remove(key);
}
- public void defrag(ChannelMessage msg ) {
+ public void defrag(ChannelMessage msg) {
FragKey key = new FragKey(msg.getUniqueId());
- FragCollection coll = getFragCollection(key,msg);
- coll.addMessage((ChannelMessage)msg.deepclone());
+ FragCollection coll = getFragCollection(key, msg);
+ coll.addMessage((ChannelMessage) msg.deepclone());
- if ( coll.complete() ) {
+ if (coll.complete()) {
removeFragCollection(key);
ChannelMessage complete = coll.assemble();
super.messageReceived(complete);
@@ -108,24 +109,24 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements
public void frag(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
int size = msg.getMessage().getLength();
- int count = ((size / maxSize )+(size%maxSize==0?0:1));
+ int count = ((size / maxSize) + (size % maxSize == 0 ? 0 : 1));
ChannelMessage[] messages = new ChannelMessage[count];
int remaining = size;
- for ( int i=0; i<count; i++ ) {
- ChannelMessage tmp = (ChannelMessage)msg.clone();
- int offset = (i*maxSize);
- int length = Math.min(remaining,maxSize);
+ for (int i = 0; i < count; i++) {
+ ChannelMessage tmp = (ChannelMessage) msg.clone();
+ int offset = (i * maxSize);
+ int length = Math.min(remaining, maxSize);
tmp.getMessage().clear();
- tmp.getMessage().append(msg.getMessage().getBytesDirect(),offset,length);
- //add the msg nr
- //tmp.getMessage().append(XByteBuffer.toBytes(i),0,4);
+ tmp.getMessage().append(msg.getMessage().getBytesDirect(), offset, length);
+ // add the msg nr
+ // tmp.getMessage().append(XByteBuffer.toBytes(i),0,4);
tmp.getMessage().append(i);
- //add the total nr of messages
- //tmp.getMessage().append(XByteBuffer.toBytes(count),0,4);
+ // add the total nr of messages
+ // tmp.getMessage().append(XByteBuffer.toBytes(count),0,4);
tmp.getMessage().append(count);
- //add true as the frag flag
- //byte[] flag = XByteBuffer.toBytes(true);
- //tmp.getMessage().append(flag,0,flag.length);
+ // add true as the frag flag
+ // byte[] flag = XByteBuffer.toBytes(true);
+ // tmp.getMessage().append(flag,0,flag.length);
tmp.getMessage().append(true);
messages[i] = tmp;
remaining -= length;
@@ -147,9 +148,9 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements
removeFragCollection(key);
}
}
- }catch ( Exception x ) {
- if ( log.isErrorEnabled() ) {
- log.error(sm.getString("fragmentationInterceptor.heartbeat.failed"),x);
+ } catch (Exception x) {
+ if (log.isErrorEnabled()) {
+ log.error(sm.getString("fragmentationInterceptor.heartbeat.failed"), x);
}
}
super.heartbeat();
@@ -179,19 +180,20 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements
private final long received = System.currentTimeMillis();
private final ChannelMessage msg;
private final XByteBuffer[] frags;
+
public FragCollection(ChannelMessage msg) {
- //get the total messages
- int count = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
+ // get the total messages
+ int count = XByteBuffer.toInt(msg.getMessage().getBytesDirect(), msg.getMessage().getLength() - 4);
frags = new XByteBuffer[count];
this.msg = msg;
}
public void addMessage(ChannelMessage msg) {
- //remove the total messages
+ // remove the total messages
msg.getMessage().trim(4);
- //get the msg nr
- int nr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
- //remove the msg nr
+ // get the msg nr
+ int nr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(), msg.getMessage().getLength() - 4);
+ // remove the msg nr
msg.getMessage().trim(4);
frags[nr] = msg.getMessage();
@@ -199,21 +201,21 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements
public boolean complete() {
boolean result = true;
- for ( int i=0; (i<frags.length) && (result); i++ ) {
+ for (int i = 0; (i < frags.length) && (result); i++) {
result = (frags[i] != null);
}
return result;
}
public ChannelMessage assemble() {
- if ( !complete() ) {
+ if (!complete()) {
throw new IllegalStateException(sm.getString("fragmentationInterceptor.fragments.missing"));
}
int buffersize = 0;
for (XByteBuffer frag : frags) {
buffersize += frag.getLength();
}
- XByteBuffer buf = new XByteBuffer(buffersize,false);
+ XByteBuffer buf = new XByteBuffer(buffersize, false);
msg.setMessage(buf);
for (XByteBuffer frag : frags) {
msg.getMessage().append(frag.getBytesDirect(), 0, frag.getLength());
@@ -222,33 +224,35 @@ public class FragmentationInterceptor extends ChannelInterceptorBase implements
}
public boolean expired(long expire) {
- return (System.currentTimeMillis()-received)>expire;
+ return (System.currentTimeMillis() - received) > expire;
}
}
public static class FragKey {
private final byte[] uniqueId;
private final long received = System.currentTimeMillis();
- public FragKey(byte[] id ) {
+
+ public FragKey(byte[] id) {
this.uniqueId = id;
}
+
@Override
public int hashCode() {
- return XByteBuffer.toInt(uniqueId,0);
+ return XByteBuffer.toInt(uniqueId, 0);
}
@Override
- public boolean equals(Object o ) {
- if ( o instanceof FragKey ) {
- return Arrays.equals(uniqueId,((FragKey)o).uniqueId);
- } else {
+ public boolean equals(Object o) {
+ if (o instanceof FragKey) {
+ return Arrays.equals(uniqueId, ((FragKey) o).uniqueId);
+ } else {
return false;
}
}
public boolean expired(long expire) {
- return (System.currentTimeMillis()-received)>expire;
+ return (System.currentTimeMillis() - received) > expire;
}
}
diff --git a/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
index 1de0dc1e38..f4d17cec0d 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java
@@ -81,20 +81,20 @@ public class GzipInterceptor extends ChannelInterceptorBase implements GzipInter
countCompressedTX.incrementAndGet();
compressedSizeTX.addAndGet(data.length);
}
- } else if (statsEnabled){
+ } else if (statsEnabled) {
countUncompressedTX.incrementAndGet();
uncompressedSizeTX.addAndGet(data.length);
}
msg.getMessage().trim(msg.getMessage().getLength());
- msg.getMessage().append(data,0,data.length);
+ msg.getMessage().append(data, 0, data.length);
super.sendMessage(destination, msg, payload);
int currentCount = count.incrementAndGet();
if (statsEnabled && interval > 0 && currentCount % interval == 0) {
report();
}
- } catch ( IOException x ) {
+ } catch (IOException x) {
log.error(sm.getString("gzipInterceptor.compress.failed"));
throw new ChannelException(x);
}
@@ -122,15 +122,15 @@ public class GzipInterceptor extends ChannelInterceptorBase implements GzipInter
}
msg.getMessage().trim(msg.getMessage().getLength());
- msg.getMessage().append(data,0,data.length);
+ msg.getMessage().append(data, 0, data.length);
super.messageReceived(msg);
int currentCount = count.incrementAndGet();
if (statsEnabled && interval > 0 && currentCount % interval == 0) {
report();
}
- } catch ( IOException x ) {
- log.error(sm.getString("gzipInterceptor.decompress.failed"),x);
+ } catch (IOException x) {
+ log.error(sm.getString("gzipInterceptor.decompress.failed"), x);
}
}
@@ -146,13 +146,14 @@ public class GzipInterceptor extends ChannelInterceptorBase implements GzipInter
/**
- * @param data Data to decompress
- * @return Decompressed data
+ * @param data Data to decompress
+ *
+ * @return Decompressed data
+ *
* @throws IOException Compression error
*/
public static byte[] decompress(byte[] data) throws IOException {
- ByteArrayOutputStream bout =
- new ByteArrayOutputStream(DEFAULT_BUFFER_SIZE);
+ ByteArrayOutputStream bout = new ByteArrayOutputStream(DEFAULT_BUFFER_SIZE);
ByteArrayInputStream bin = new ByteArrayInputStream(data);
GZIPInputStream gin = new GZIPInputStream(bin);
byte[] tmp = new byte[DEFAULT_BUFFER_SIZE];
@@ -170,10 +171,8 @@ public class GzipInterceptor extends ChannelInterceptorBase implements GzipInter
log.info(sm.getString("gzipInterceptor.report", Integer.valueOf(getCount()),
Integer.valueOf(getCountCompressedTX()), Integer.valueOf(getCountUncompressedTX()),
Integer.valueOf(getCountCompressedRX()), Integer.valueOf(getCountUncompressedRX()),
- Long.valueOf(getSizeTX()), Long.valueOf(getCompressedSizeTX()),
- Long.valueOf(getUncompressedSizeTX()),
- Long.valueOf(getSizeRX()), Long.valueOf(getCompressedSizeRX()),
- Long.valueOf(getUncompressedSizeRX())));
+ Long.valueOf(getSizeTX()), Long.valueOf(getCompressedSizeTX()), Long.valueOf(getUncompressedSizeTX()),
+ Long.valueOf(getSizeRX()), Long.valueOf(getCompressedSizeRX()), Long.valueOf(getUncompressedSizeRX())));
}
diff --git a/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptorMBean.java b/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptorMBean.java
index 07f30c01ac..762e6f6415 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptorMBean.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/GzipInterceptorMBean.java
@@ -20,43 +20,43 @@ public interface GzipInterceptorMBean {
// Config
int getOptionFlag();
+
void setOptionFlag(int optionFlag);
/**
* @return the minimum payload size for compression to be enabled.
*/
int getCompressionMinSize();
+
/**
- * Set the minimum payload size for compression to be enabled. A value of
- * zero or less means compression will always be used. If not explicitly
- * configured, a default of zero will be used.
+ * Set the minimum payload size for compression to be enabled. A value of zero or less means compression will always
+ * be used. If not explicitly configured, a default of zero will be used.
*
* @param compressionMinSize The new minimum payload size
*/
void setCompressionMinSize(int compressionMinSize);
/**
- * @return {@code true} if the interceptor is configured to collect
- * statistics, otherwise {@code false}
+ * @return {@code true} if the interceptor is configured to collect statistics, otherwise {@code false}
*/
boolean getStatsEnabled();
+
/**
* Configure whether the interceptor collects statistics.
*
- * @param statsEnabled {@code true} to enable statistics collections,
- * otherwise {@code false}
+ * @param statsEnabled {@code true} to enable statistics collections, otherwise {@code false}
*/
void setStatsEnabled(boolean statsEnabled);
/**
- * @return If statistics collection is enabled, the number of messages
- * between statistics reports being written to the log.
+ * @return If statistics collection is enabled, the number of messages between statistics reports being written to
+ * the log.
*/
int getInterval();
+
/**
- * If statistics collection is enabled, set the number of messages between
- * statistics reports being written to the log. A value of zero or less
- * means no statistics reports are written.
+ * If statistics collection is enabled, set the number of messages between statistics reports being written to the
+ * log. A value of zero or less means no statistics reports are written.
*
* @param interval The new interval between reports
*/
@@ -64,16 +64,28 @@ public interface GzipInterceptorMBean {
// Stats
int getCount();
+
int getCountCompressedTX();
+
int getCountUncompressedTX();
+
int getCountCompressedRX();
+
int getCountUncompressedRX();
+
long getSizeTX();
+
long getCompressedSizeTX();
+
long getUncompressedSizeTX();
+
long getSizeRX();
+
long getCompressedSizeRX();
+
long getUncompressedSizeRX();
+
void reset();
+
void report();
}
diff --git a/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
index 23b4ad86ec..642bc0a955 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java
@@ -36,19 +36,16 @@ import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
/**
- * The message dispatcher is a way to enable asynchronous communication
- * through a channel. The dispatcher will look for the
- * <code>Channel.SEND_OPTIONS_ASYNCHRONOUS</code> flag to be set, if it is, it
- * will queue the message for delivery and immediately return to the sender.
+ * The message dispatcher is a way to enable asynchronous communication through a channel. The dispatcher will look for
+ * the <code>Channel.SEND_OPTIONS_ASYNCHRONOUS</code> flag to be set, if it is, it will queue the message for delivery
+ * and immediately return to the sender.
*/
-public class MessageDispatchInterceptor extends ChannelInterceptorBase
- implements MessageDispatchInterceptorMBean {
+public class MessageDispatchInterceptor extends ChannelInterceptorBase implements MessageDispatchInterceptorMBean {
private static final Log log = LogFactory.getLog(MessageDispatchInterceptor.class);
- protected static final StringManager sm =
- StringManager.getManager(MessageDispatchInterceptor.class);
+ protected static final StringManager sm = StringManager.getManager(MessageDispatchInterceptor.class);
- protected long maxQueueSize = 1024*1024*64; //64 MiB
+ protected long maxQueueSize = 1024 * 1024 * 64; // 64 MiB
protected volatile boolean run = false;
protected boolean useDeepClone = true;
protected boolean alwaysSend = true;
@@ -68,25 +65,23 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
throws ChannelException {
- boolean async = (msg.getOptions() &
- Channel.SEND_OPTIONS_ASYNCHRONOUS) == Channel.SEND_OPTIONS_ASYNCHRONOUS;
+ boolean async = (msg.getOptions() & Channel.SEND_OPTIONS_ASYNCHRONOUS) == Channel.SEND_OPTIONS_ASYNCHRONOUS;
if (async && run) {
- if ((getCurrentSize()+msg.getMessage().getLength()) > maxQueueSize) {
+ if ((getCurrentSize() + msg.getMessage().getLength()) > maxQueueSize) {
if (alwaysSend) {
- super.sendMessage(destination,msg,payload);
+ super.sendMessage(destination, msg, payload);
return;
} else {
throw new ChannelException(sm.getString("messageDispatchInterceptor.queue.full",
Long.toString(maxQueueSize), Long.toString(getCurrentSize())));
}
}
- //add to queue
+ // add to queue
if (useDeepClone) {
- msg = (ChannelMessage)msg.deepclone();
+ msg = (ChannelMessage) msg.deepclone();
}
if (!addToQueue(msg, destination, payload)) {
- throw new ChannelException(
- sm.getString("messageDispatchInterceptor.unableAdd.queue"));
+ throw new ChannelException(sm.getString("messageDispatchInterceptor.unableAdd.queue"));
}
addAndGetCurrentSize(msg.getMessage().getLength());
} else {
@@ -95,8 +90,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
}
- public boolean addToQueue(final ChannelMessage msg, final Member[] destination,
- final InterceptorPayload payload) {
+ public boolean addToQueue(final ChannelMessage msg, final Member[] destination, final InterceptorPayload payload) {
executor.execute(() -> sendAsyncData(msg, destination, payload));
return true;
}
@@ -110,8 +104,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
if (getChannel().getName() != null) {
channelName = "[" + getChannel().getName() + "]";
}
- executor = ExecutorFactory.newThreadPool(maxSpareThreads, maxThreads, keepAliveTime,
- TimeUnit.MILLISECONDS,
+ executor = ExecutorFactory.newThreadPool(maxSpareThreads, maxThreads, keepAliveTime, TimeUnit.MILLISECONDS,
new TcclThreadFactory("MessageDispatchInterceptor.MessageDispatchThread" + channelName));
run = true;
}
@@ -126,7 +119,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
@Override
public void setOptionFlag(int flag) {
- if ( flag != Channel.SEND_OPTIONS_ASYNCHRONOUS ) {
+ if (flag != Channel.SEND_OPTIONS_ASYNCHRONOUS) {
log.warn(sm.getString("messageDispatchInterceptor.warning.optionflag"));
}
super.setOptionFlag(flag);
@@ -212,11 +205,11 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
@Override
public void start(int svc) throws ChannelException {
- //start the thread
- if (!run ) {
+ // start the thread
+ if (!run) {
synchronized (this) {
// only start with the sender
- if ( !run && ((svc & Channel.SND_TX_SEQ)==Channel.SND_TX_SEQ) ) {
+ if (!run && ((svc & Channel.SND_TX_SEQ) == Channel.SND_TX_SEQ)) {
startQueue();
}
}
@@ -227,10 +220,10 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
@Override
public void stop(int svc) throws ChannelException {
- //stop the thread
+ // stop the thread
if (run) {
synchronized (this) {
- if ( run && ((svc & Channel.SND_TX_SEQ)==Channel.SND_TX_SEQ)) {
+ if (run && ((svc & Channel.SND_TX_SEQ) == Channel.SND_TX_SEQ)) {
stopQueue();
}
}
@@ -240,8 +233,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
}
- protected void sendAsyncData(ChannelMessage msg, Member[] destination,
- InterceptorPayload payload) {
+ protected void sendAsyncData(ChannelMessage msg, Member[] destination, InterceptorPayload payload) {
ErrorHandler handler = null;
if (payload != null) {
handler = payload.getErrorHandler();
@@ -252,10 +244,10 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
if (handler != null) {
handler.handleCompletion(new UniqueId(msg.getUniqueId()));
}
- } catch ( Exception ex ) {
- log.error(sm.getString("messageDispatchInterceptor.completeMessage.failed"),ex);
+ } catch (Exception ex) {
+ log.error(sm.getString("messageDispatchInterceptor.completeMessage.failed"), ex);
}
- } catch ( Exception x ) {
+ } catch (Exception x) {
ChannelException cx = null;
if (x instanceof ChannelException) {
cx = (ChannelException) x;
@@ -263,14 +255,14 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
cx = new ChannelException(x);
}
if (log.isDebugEnabled()) {
- log.debug(sm.getString("messageDispatchInterceptor.AsyncMessage.failed"),x);
+ log.debug(sm.getString("messageDispatchInterceptor.AsyncMessage.failed"), x);
}
try {
if (handler != null) {
handler.handleError(cx, new UniqueId(msg.getUniqueId()));
}
- } catch ( Exception ex ) {
- log.error(sm.getString("messageDispatchInterceptor.errorMessage.failed"),ex);
+ } catch (Exception ex) {
+ log.error(sm.getString("messageDispatchInterceptor.errorMessage.failed"), ex);
}
} finally {
addAndGetCurrentSize(-msg.getMessage().getLength());
@@ -280,6 +272,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
// ---------------------------------------------- stats of the thread pool
/**
* Return the current number of threads that are managed by the pool.
+ *
* @return the current number of threads that are managed by the pool
*/
@Override
@@ -293,6 +286,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
/**
* Return the current number of threads that are in use.
+ *
* @return the current number of threads that are in use
*/
@Override
@@ -306,6 +300,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
/**
* Return the total number of tasks that have ever been scheduled for execution by the pool.
+ *
* @return the total number of tasks that have ever been scheduled for execution by the pool
*/
@Override
@@ -319,6 +314,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase
/**
* Return the total number of tasks that have completed execution by the pool.
+ *
* @return the total number of tasks that have completed execution by the pool
*/
@Override
diff --git a/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java b/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
index 5aedf1f06b..be6f6f3c9a 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/NonBlockingCoordinator.java
@@ -43,34 +43,35 @@ import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
/**
- * <p>Title: Auto merging leader election algorithm</p>
- *
- * <p>Description: Implementation of a simple coordinator algorithm that not only selects a coordinator,
- * it also merges groups automatically when members are discovered that weren't part of the
- * </p>
- * <p>This algorithm is non blocking meaning it allows for transactions while the coordination phase is going on
+ * <p>
+ * Title: Auto merging leader election algorithm
+ * </p>
+ * <p>
+ * Description: Implementation of a simple coordinator algorithm that not only selects a coordinator, it also merges
+ * groups automatically when members are discovered that weren't part of the
+ * </p>
+ * <p>
+ * This algorithm is non blocking meaning it allows for transactions while the coordination phase is going on
* </p>
- * <p>This implementation is based on a home brewed algorithm that uses the AbsoluteOrder of a membership
- * to pass a token ring of the current membership.<br>
+ * <p>
+ * This implementation is based on a home brewed algorithm that uses the AbsoluteOrder of a membership to pass a token
+ * ring of the current membership.<br>
* This is not the same as just using AbsoluteOrder! Consider the following scenario:<br>
- * Nodes, A,B,C,D,E on a network, in that priority. AbsoluteOrder will only work if all
- * nodes are receiving pings from all the other nodes.
- * meaning, that node{i} receives pings from node{all}-node{i}<br>
- * but the following could happen if a multicast problem occurs.
- * A has members {B,C,D}<br>
+ * Nodes, A,B,C,D,E on a network, in that priority. AbsoluteOrder will only work if all nodes are receiving pings from
+ * all the other nodes. meaning, that node{i} receives pings from node{all}-node{i}<br>
+ * but the following could happen if a multicast problem occurs. A has members {B,C,D}<br>
* B has members {A,C}<br>
* C has members {D,E}<br>
* D has members {A,B,C,E}<br>
* E has members {A,C,D}<br>
- * Because the default Tribes membership implementation, relies on the multicast packets to
- * arrive at all nodes correctly, there is nothing guaranteeing that it will.<br>
+ * Because the default Tribes membership implementation, relies on the multicast packets to arrive at all nodes
+ * correctly, there is nothing guaranteeing that it will.<br>
* <br>
- * To best explain how this algorithm works, lets take the above example:
- * For simplicity we assume that a send operation is O(1) for all nodes, although this algorithm will work
- * where messages overlap, as they all depend on absolute order<br>
- * Scenario 1: A,B,C,D,E all come online at the same time
- * Eval phase, A thinks of itself as leader, B thinks of A as leader,
- * C thinks of itself as leader, D,E think of A as leader<br>
+ * To best explain how this algorithm works, lets take the above example: For simplicity we assume that a send operation
+ * is O(1) for all nodes, although this algorithm will work where messages overlap, as they all depend on absolute
+ * order<br>
+ * Scenario 1: A,B,C,D,E all come online at the same time Eval phase, A thinks of itself as leader, B thinks of A as
+ * leader, C thinks of itself as leader, D,E think of A as leader<br>
* Token phase:<br>
* (1) A sends out a message X{A-ldr, A-src, mbrs-A,B,C,D} to B where X is the id for the message(and the view)<br>
* (1) C sends out a message Y{C-ldr, C-src, mbrs-C,D,E} to D where Y is the id for the message(and the view)<br>
@@ -88,15 +89,15 @@ import org.apache.juli.logging.LogFactory;
* D - {A-ldr, mbrs-A,B,C,D,E, id=X}<br>
* E - {A-ldr, mbrs-A,B,C,D,E, id=Y}<br>
* <br>
- * A message doesn't stop until it reaches its original sender, unless its dropped by a higher leader.
- * As you can see, E still thinks the viewId=Y, which is not correct. But at this point we have
- * arrived at the same membership and all nodes are informed of each other.<br>
+ * A message doesn't stop until it reaches its original sender, unless its dropped by a higher leader. As you can see, E
+ * still thinks the viewId=Y, which is not correct. But at this point we have arrived at the same membership and all
+ * nodes are informed of each other.<br>
* To synchronize the rest we simply perform the following check at A when A receives X:<br>
* Original X{A-ldr, A-src, mbrs-A,B,C,D} == Arrived X{A-ldr, A-src, mbrs-A,B,C,D,E}<br>
- * Since the condition is false, A, will resend the token, and A sends X{A-ldr, A-src, mbrs-A,B,C,D,E} to B
- * When A receives X again, the token is complete. <br>
- * Optionally, A can send a message X{A-ldr, A-src, mbrs-A,B,C,D,E confirmed} to A,B,C,D,E who then
- * install and accept the view.
+ * Since the condition is false, A, will resend the token, and A sends X{A-ldr, A-src, mbrs-A,B,C,D,E} to B When A
+ * receives X again, the token is complete. <br>
+ * Optionally, A can send a message X{A-ldr, A-src, mbrs-A,B,C,D,E confirmed} to A,B,C,D,E who then install and accept
+ * the view.
* </p>
* <p>
* Lets assume that C1 arrives, C1 has lower priority than C, but higher priority than D.<br>
@@ -106,20 +107,27 @@ import org.apache.juli.logging.LogFactory;
* In this case, C1 sends a Z{C1-ldr, C1-src, mbrs-C1,D,E} to D<br>
* D receives Z{C1-ldr, C1-src, mbrs-C1,D,E} and sends Z{A-ldr, C1-src, mbrs-A,B,C,C1,D,E} to E<br>
* E receives Z{A-ldr, C1-src, mbrs-A,B,C,C1,D,E} and sends it to A<br>
- * A sends Z{A-ldr, A-src, mbrs-A,B,C,C1,D,E} to B and the chain continues until A receives the token again.
- * At that time A optionally sends out Z{A-ldr, A-src, mbrs-A,B,C,C1,D,E, confirmed} to A,B,C,C1,D,E
+ * A sends Z{A-ldr, A-src, mbrs-A,B,C,C1,D,E} to B and the chain continues until A receives the token again. At that
+ * time A optionally sends out Z{A-ldr, A-src, mbrs-A,B,C,C1,D,E, confirmed} to A,B,C,C1,D,E
* </p>
- * <p>To ensure that the view gets implemented at all nodes at the same time,
- * A will send out a VIEW_CONF message, this is the 'confirmed' message that is optional above.
- * <p>Ideally, the interceptor below this one would be the TcpFailureDetector to ensure correct memberships</p>
- *
- * <p>The example above, of course can be simplified with a finite statemachine:<br>
+ * <p>
+ * To ensure that the view gets implemented at all nodes at the same time, A will send out a VIEW_CONF message, this is
+ * the 'confirmed' message that is optional above.
+ * <p>
+ * Ideally, the interceptor below this one would be the TcpFailureDetector to ensure correct memberships
+ * </p>
+ * <p>
+ * The example above, of course can be simplified with a finite statemachine:<br>
* But I suck at writing state machines, my head gets all confused. One day I will document this algorithm though.<br>
* Maybe I'll do a state diagram :)
* </p>
* <h2>State Diagrams</h2>
- * <a href="https://people.apache.org/~fhanik/tribes/docs/leader-election-initiate-election.jpg">Initiate an election</a><br><br>
- * <a href="https://people.apache.org/~fhanik/tribes/docs/leader-election-message-arrives.jpg">Receive an election message</a><br><br>
+ * <a href="https://people.apache.org/~fhanik/tribes/docs/leader-election-initiate-election.jpg">Initiate an
+ * election</a><br>
+ * <br>
+ * <a href="https://people.apache.org/~fhanik/tribes/docs/leader-election-message-arrives.jpg">Receive an election
+ * message</a><br>
+ * <br>
*/
public class NonBlockingCoordinator extends ChannelInterceptorBase {
@@ -129,23 +137,26 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
/**
* header for a coordination message
*/
- protected static final byte[] COORD_HEADER = new byte[] {-86, 38, -34, -29, -98, 90, 65, 63, -81, -122, -6, -110, 99, -54, 13, 63};
+ protected static final byte[] COORD_HEADER =
+ new byte[] { -86, 38, -34, -29, -98, 90, 65, 63, -81, -122, -6, -110, 99, -54, 13, 63 };
/**
* Coordination request
*/
- protected static final byte[] COORD_REQUEST = new byte[] {104, -95, -92, -42, 114, -36, 71, -19, -79, 20, 122, 101, -1, -48, -49, 30};
+ protected static final byte[] COORD_REQUEST =
+ new byte[] { 104, -95, -92, -42, 114, -36, 71, -19, -79, 20, 122, 101, -1, -48, -49, 30 };
/**
* Coordination confirmation, for blocking installations
*/
- protected static final byte[] COORD_CONF = new byte[] {67, 88, 107, -86, 69, 23, 76, -70, -91, -23, -87, -25, -125, 86, 75, 20};
+ protected static final byte[] COORD_CONF =
+ new byte[] { 67, 88, 107, -86, 69, 23, 76, -70, -91, -23, -87, -25, -125, 86, 75, 20 };
/**
* Alive message
*/
- protected static final byte[] COORD_ALIVE = new byte[] {79, -121, -25, -15, -59, 5, 64, 94, -77, 113, -119, -88, 52, 114, -56, -46,
- -18, 102, 10, 34, -127, -9, 71, 115, -70, 72, -101, 88, 72, -124, 127, 111,
- 74, 76, -116, 50, 111, 103, 65, 3, -77, 51, -35, 0, 119, 117, 9, -26,
- 119, 50, -75, -105, -102, 36, 79, 37, -68, -84, -123, 15, -22, -109, 106, -55};
+ protected static final byte[] COORD_ALIVE = new byte[] { 79, -121, -25, -15, -59, 5, 64, 94, -77, 113, -119, -88,
+ 52, 114, -56, -46, -18, 102, 10, 34, -127, -9, 71, 115, -70, 72, -101, 88, 72, -124, 127, 111, 74, 76, -116,
+ 50, 111, 103, 65, 3, -77, 51, -35, 0, 119, 117, 9, -26, 119, 50, -75, -105, -102, 36, 79, 37, -68, -84,
+ -123, 15, -22, -109, 106, -55 };
/**
* Time to wait for coordination timeout
*/
@@ -165,8 +176,7 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
protected Membership membership = null;
/**
- * indicates that we are running an election
- * and this is the one we are running
+ * indicates that we are running an election and this is the one we are running
*/
protected UniqueId suggestedviewId;
protected volatile Membership suggestedView;
@@ -182,49 +192,56 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
super();
}
-//============================================================================================================
-// COORDINATION HANDLING
-//============================================================================================================
+ // ============================================================================================================
+ // COORDINATION HANDLING
+ // ============================================================================================================
public void startElection(boolean force) throws ChannelException {
synchronized (electionMutex) {
Member local = getLocalMember(false);
Member[] others = membership.getMembers();
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_START_ELECT,this,"Election initiated"));
- if ( others.length == 0 ) {
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_START_ELECT, this, "Election initiated"));
+ if (others.length == 0) {
this.viewId = new UniqueId(UUIDGenerator.randomUUID(false));
- this.view = new Membership(local,AbsoluteOrder.comp, true);
- this.handleViewConf(createElectionMsg(local,others,local), view);
- return; //the only member, no need for an election
+ this.view = new Membership(local, AbsoluteOrder.comp, true);
+ this.handleViewConf(createElectionMsg(local, others, local), view);
+ return; // the only member, no need for an election
}
- if ( suggestedviewId != null ) {
+ if (suggestedviewId != null) {
- if ( view != null && Arrays.diff(view,suggestedView,local).length == 0 && Arrays.diff(suggestedView,view,local).length == 0) {
+ if (view != null && Arrays.diff(view, suggestedView, local).length == 0 &&
+ Arrays.diff(suggestedView, view, local).length == 0) {
suggestedviewId = null;
suggestedView = null;
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_ELECT_ABANDONED,this,"Election abandoned, running election matches view"));
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_ELECT_ABANDONED, this,
+ "Election abandoned, running election matches view"));
} else {
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_ELECT_ABANDONED,this,"Election abandoned, election running"));
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_ELECT_ABANDONED, this,
+ "Election abandoned, election running"));
}
- return; //election already running, I'm not allowed to have two of them
+ return; // election already running, I'm not allowed to have two of them
}
- if ( view != null && Arrays.diff(view,membership,local).length == 0 && Arrays.diff(membership,view,local).length == 0) {
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_ELECT_ABANDONED,this,"Election abandoned, view matches membership"));
- return; //already have this view installed
+ if (view != null && Arrays.diff(view, membership, local).length == 0 &&
+ Arrays.diff(membership, view, local).length == 0) {
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_ELECT_ABANDONED, this,
+ "Election abandoned, view matches membership"));
+ return; // already have this view installed
}
- int prio = AbsoluteOrder.comp.compare(local,others[0]);
- Member leader = ( prio < 0 )?local:others[0];//am I the leader in my view?
- if ( local.equals(leader) || force ) {
+ int prio = AbsoluteOrder.comp.compare(local, others[0]);
+ Member leader = (prio < 0) ? local : others[0];// am I the leader in my view?
+ if (local.equals(leader) || force) {
CoordinationMessage msg = createElectionMsg(local, others, leader);
suggestedviewId = msg.getId();
- suggestedView = new Membership(local,AbsoluteOrder.comp,true);
- Arrays.fill(suggestedView,msg.getMembers());
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_PROCESS_ELECT,this,"Election, sending request"));
- sendElectionMsg(local,others[0],msg);
+ suggestedView = new Membership(local, AbsoluteOrder.comp, true);
+ Arrays.fill(suggestedView, msg.getMembers());
+ fireInterceptorEvent(
+ new CoordinationEvent(CoordinationEvent.EVT_PROCESS_ELECT, this, "Election, sending request"));
+ sendElectionMsg(local, others[0], msg);
} else {
try {
coordMsgReceived.set(false);
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_WAIT_FOR_MSG,this,"Election, waiting for request"));
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_WAIT_FOR_MSG, this,
+ "Election, waiting for request"));
electionMutex.wait(waitForCoordMsgTimeout);
} catch (InterruptedException x) {
Thread.currentThread().interrupt();
@@ -239,39 +256,40 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
} else {
msg = "Election abandoned, received a message";
}
- fireInterceptorEvent(new CoordinationEvent(
- CoordinationEvent.EVT_ELECT_ABANDONED, this, msg));
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_ELECT_ABANDONED, this, msg));
}
}
}
private CoordinationMessage createElectionMsg(Member local, Member[] others, Member leader) {
- Membership m = new Membership(local,AbsoluteOrder.comp,true);
- Arrays.fill(m,others);
+ Membership m = new Membership(local, AbsoluteOrder.comp, true);
+ Arrays.fill(m, others);
Member[] mbrs = m.getMembers();
m.reset();
- CoordinationMessage msg = new CoordinationMessage(leader, local, mbrs,new UniqueId(UUIDGenerator.randomUUID(true)), COORD_REQUEST);
+ CoordinationMessage msg = new CoordinationMessage(leader, local, mbrs,
+ new UniqueId(UUIDGenerator.randomUUID(true)), COORD_REQUEST);
return msg;
}
protected void sendElectionMsg(Member local, Member next, CoordinationMessage msg) throws ChannelException {
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_SEND_MSG,this,"Sending election message to("+next.getName()+")"));
- super.sendMessage(new Member[] {next}, createData(msg, local), null);
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_SEND_MSG, this,
+ "Sending election message to(" + next.getName() + ")"));
+ super.sendMessage(new Member[] { next }, createData(msg, local), null);
}
protected void sendElectionMsgToNextInline(Member local, CoordinationMessage msg) throws ChannelException {
- int next = Arrays.nextIndex(local,msg.getMembers());
+ int next = Arrays.nextIndex(local, msg.getMembers());
int current = next;
msg.leader = msg.getMembers()[0];
- boolean sent = false;
- while ( !sent && current >= 0 ) {
+ boolean sent = false;
+ while (!sent && current >= 0) {
try {
sendElectionMsg(local, msg.getMembers()[current], msg);
sent = true;
- }catch ( ChannelException x ) {
+ } catch (ChannelException x) {
log.warn(sm.getString("nonBlockingCoordinator.electionMessage.sendfailed", msg.getMembers()[current]));
- current = Arrays.nextIndex(msg.getMembers()[current],msg.getMembers());
- if ( current == next ) {
+ current = Arrays.nextIndex(msg.getMembers()[current], msg.getMembers());
+ if (current == next) {
throw x;
}
}
@@ -293,8 +311,8 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
}
protected boolean memberAlive(Member mbr, long conTimeout) {
- //could be a shutdown notification
- if ( Arrays.equals(mbr.getCommand(),Member.SHUTDOWN_PAYLOAD) ) {
+ // could be a shutdown notification
+ if (Arrays.equals(mbr.getCommand(), Member.SHUTDOWN_PAYLOAD)) {
return false;
}
@@ -304,20 +322,20 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
socket.connect(addr, (int) conTimeout);
return true;
} catch (SocketTimeoutException | ConnectException x) {
- //do nothing, we couldn't connect
+ // do nothing, we couldn't connect
} catch (Exception x) {
- log.error(sm.getString("nonBlockingCoordinator.memberAlive.failed"),x);
+ log.error(sm.getString("nonBlockingCoordinator.memberAlive.failed"), x);
}
return false;
}
protected Membership mergeOnArrive(CoordinationMessage msg) {
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_PRE_MERGE,this,"Pre merge"));
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_PRE_MERGE, this, "Pre merge"));
Member local = getLocalMember(false);
- Membership merged = new Membership(local,AbsoluteOrder.comp,true);
- Arrays.fill(merged,msg.getMembers());
- Arrays.fill(merged,getMembers());
- Member[] diff = Arrays.diff(merged,membership,local);
+ Membership merged = new Membership(local, AbsoluteOrder.comp, true);
+ Arrays.fill(merged, msg.getMembers());
+ Arrays.fill(merged, getMembers());
+ Member[] diff = Arrays.diff(merged, membership, local);
for (Member member : diff) {
if (!alive(member)) {
merged.removeMember(member);
@@ -325,14 +343,16 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
memberAdded(member, false);
}
}
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_POST_MERGE,this,"Post merge"));
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_POST_MERGE, this, "Post merge"));
return merged;
}
protected void processCoordMessage(CoordinationMessage msg) throws ChannelException {
- if ( !coordMsgReceived.get() ) {
+ if (!coordMsgReceived.get()) {
coordMsgReceived.set(true);
- synchronized (electionMutex) { electionMutex.notifyAll();}
+ synchronized (electionMutex) {
+ electionMutex.notifyAll();
+ }
}
Membership merged = mergeOnArrive(msg);
if (isViewConf(msg)) {
@@ -344,8 +364,8 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
protected void handleToken(CoordinationMessage msg, Membership merged) throws ChannelException {
Member local = getLocalMember(false);
- if ( local.equals(msg.getSource()) ) {
- //my message msg.src=local
+ if (local.equals(msg.getSource())) {
+ // my message msg.src=local
handleMyToken(local, msg, merged);
} else {
handleOtherToken(local, msg, merged);
@@ -353,85 +373,86 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
}
protected void handleMyToken(Member local, CoordinationMessage msg, Membership merged) throws ChannelException {
- if ( local.equals(msg.getLeader()) ) {
- //no leadership change
- if ( Arrays.sameMembers(msg.getMembers(),merged.getMembers()) ) {
+ if (local.equals(msg.getLeader())) {
+ // no leadership change
+ if (Arrays.sameMembers(msg.getMembers(), merged.getMembers())) {
msg.type = COORD_CONF;
- super.sendMessage(Arrays.remove(msg.getMembers(),local),createData(msg,local),null);
+ super.sendMessage(Arrays.remove(msg.getMembers(), local), createData(msg, local), null);
handleViewConf(msg, merged);
} else {
- //membership change
- suggestedView = new Membership(local,AbsoluteOrder.comp,true);
+ // membership change
+ suggestedView = new Membership(local, AbsoluteOrder.comp, true);
suggestedviewId = msg.getId();
- Arrays.fill(suggestedView,merged.getMembers());
+ Arrays.fill(suggestedView, merged.getMembers());
msg.view = merged.getMembers();
- sendElectionMsgToNextInline(local,msg);
+ sendElectionMsgToNextInline(local, msg);
}
} else {
- //leadership change
+ // leadership change
suggestedView = null;
suggestedviewId = null;
msg.view = merged.getMembers();
- sendElectionMsgToNextInline(local,msg);
+ sendElectionMsgToNextInline(local, msg);
}
}
protected void handleOtherToken(Member local, CoordinationMessage msg, Membership merged) throws ChannelException {
- if ( local.equals(msg.getLeader()) ) {
- //I am the new leader
- //startElection(false);
+ if (local.equals(msg.getLeader())) {
+ // I am the new leader
+ // startElection(false);
} else {
msg.view = merged.getMembers();
- sendElectionMsgToNextInline(local,msg);
+ sendElectionMsgToNextInline(local, msg);
}
}
protected void handleViewConf(CoordinationMessage msg, Membership merged) throws ChannelException {
- if ( viewId != null && msg.getId().equals(viewId) )
- {
- return;//we already have this view
+ if (viewId != null && msg.getId().equals(viewId)) {
+ return;// we already have this view
}
- view = new Membership(getLocalMember(false),AbsoluteOrder.comp,true);
- Arrays.fill(view,msg.getMembers());
+ view = new Membership(getLocalMember(false), AbsoluteOrder.comp, true);
+ Arrays.fill(view, msg.getMembers());
viewId = msg.getId();
- if ( viewId.equals(suggestedviewId) ) {
+ if (viewId.equals(suggestedviewId)) {
suggestedView = null;
suggestedviewId = null;
}
- if (suggestedView != null && AbsoluteOrder.comp.compare(suggestedView.getMembers()[0],merged.getMembers()[0])<0 ) {
+ if (suggestedView != null &&
+ AbsoluteOrder.comp.compare(suggestedView.getMembers()[0], merged.getMembers()[0]) < 0) {
suggestedView = null;
suggestedviewId = null;
}
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_CONF_RX,this,"Accepted View"));
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_CONF_RX, this, "Accepted View"));
- if ( suggestedviewId == null && hasHigherPriority(merged.getMembers(),membership.getMembers()) ) {
+ if (suggestedviewId == null && hasHigherPriority(merged.getMembers(), membership.getMembers())) {
startElection(false);
}
}
protected boolean isViewConf(CoordinationMessage msg) {
- return Arrays.contains(msg.getType(),0,COORD_CONF,0,COORD_CONF.length);
+ return Arrays.contains(msg.getType(), 0, COORD_CONF, 0, COORD_CONF.length);
}
protected boolean hasHigherPriority(Member[] complete, Member[] local) {
- if ( local == null || local.length == 0 ) {
+ if (local == null || local.length == 0) {
return false;
}
- if ( complete == null || complete.length == 0 ) {
+ if (complete == null || complete.length == 0) {
return true;
}
AbsoluteOrder.absoluteOrder(complete);
AbsoluteOrder.absoluteOrder(local);
- return (AbsoluteOrder.comp.compare(complete[0],local[0]) > 0);
+ return (AbsoluteOrder.comp.compare(complete[0], local[0]) > 0);
}
/**
* Returns coordinator if one is available
+ *
* @return Member
*/
public Member getCoordinator() {
@@ -447,46 +468,46 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
}
/**
- * Block in/out messages while a election is going on
- */
- protected void halt() {
+ * Block in/out messages while a election is going on
+ */
+ protected void halt() {
- }
+ }
- /**
- * Release lock for in/out messages election is completed
- */
- protected void release() {
+ /**
+ * Release lock for in/out messages election is completed
+ */
+ protected void release() {
- }
+ }
- /**
- * Wait for an election to end
- */
- protected void waitForRelease() {
+ /**
+ * Wait for an election to end
+ */
+ protected void waitForRelease() {
- }
+ }
-//============================================================================================================
-// OVERRIDDEN METHODS FROM CHANNEL INTERCEPTOR BASE
-//============================================================================================================
+ // ============================================================================================================
+ // OVERRIDDEN METHODS FROM CHANNEL INTERCEPTOR BASE
+ // ============================================================================================================
@Override
public void start(int svc) throws ChannelException {
- if (membership == null) {
- setupMembership();
- }
- if (started) {
- return;
- }
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_START, this, "Before start"));
- super.start(startsvc);
- started = true;
- if (view == null) {
- view = new Membership(super.getLocalMember(true), AbsoluteOrder.comp, true);
- }
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_START, this, "After start"));
- startElection(false);
+ if (membership == null) {
+ setupMembership();
+ }
+ if (started) {
+ return;
+ }
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_START, this, "Before start"));
+ super.start(startsvc);
+ started = true;
+ if (view == null) {
+ view = new Membership(super.getLocalMember(true), AbsoluteOrder.comp, true);
+ }
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_START, this, "After start"));
+ startElection(false);
}
@Override
@@ -507,32 +528,34 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
this.membership.reset();
fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_STOP, this, "After stop"));
}
- }finally {
+ } finally {
release();
}
}
@Override
- public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
+ public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
+ throws ChannelException {
waitForRelease();
super.sendMessage(destination, msg, payload);
}
@Override
public void messageReceived(ChannelMessage msg) {
- if ( Arrays.contains(msg.getMessage().getBytesDirect(),0,COORD_ALIVE,0,COORD_ALIVE.length) ) {
- //ignore message, its an alive message
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MSG_ARRIVE,this,"Alive Message"));
+ if (Arrays.contains(msg.getMessage().getBytesDirect(), 0, COORD_ALIVE, 0, COORD_ALIVE.length)) {
+ // ignore message, its an alive message
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MSG_ARRIVE, this, "Alive Message"));
- } else if ( Arrays.contains(msg.getMessage().getBytesDirect(),0,COORD_HEADER,0,COORD_HEADER.length) ) {
+ } else if (Arrays.contains(msg.getMessage().getBytesDirect(), 0, COORD_HEADER, 0, COORD_HEADER.length)) {
try {
CoordinationMessage cmsg = new CoordinationMessage(msg.getMessage());
Member[] cmbr = cmsg.getMembers();
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MSG_ARRIVE,this,"Coord Msg Arrived("+Arrays.toNameString(cmbr)+")"));
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MSG_ARRIVE, this,
+ "Coord Msg Arrived(" + Arrays.toNameString(cmbr) + ")"));
processCoordMessage(cmsg);
- }catch ( ChannelException x ) {
- log.error(sm.getString("nonBlockingCoordinator.processCoordinationMessage.failed"),x);
+ } catch (ChannelException x) {
+ log.error(sm.getString("nonBlockingCoordinator.processCoordinationMessage.failed"), x);
}
} else {
super.messageReceived(msg);
@@ -541,10 +564,10 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
@Override
public void memberAdded(Member member) {
- memberAdded(member,true);
+ memberAdded(member, true);
}
- public void memberAdded(Member member,boolean elect) {
+ public void memberAdded(Member member, boolean elect) {
if (membership == null) {
setupMembership();
}
@@ -552,12 +575,13 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
super.memberAdded(member);
}
try {
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MBR_ADD,this,"Member add("+member.getName()+")"));
+ fireInterceptorEvent(
+ new CoordinationEvent(CoordinationEvent.EVT_MBR_ADD, this, "Member add(" + member.getName() + ")"));
if (started && elect) {
startElection(false);
}
} catch (ChannelException x) {
- log.error(sm.getString("nonBlockingCoordinator.memberAdded.failed"),x);
+ log.error(sm.getString("nonBlockingCoordinator.memberAdded.failed"), x);
}
}
@@ -566,22 +590,22 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
membership.removeMember(member);
super.memberDisappeared(member);
try {
- fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MBR_DEL,this,"Member remove("+member.getName()+")"));
- if (started && (isCoordinator() || isHighest()))
- {
- startElection(true); //to do, if a member disappears, only the coordinator can start
+ fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MBR_DEL, this,
+ "Member remove(" + member.getName() + ")"));
+ if (started && (isCoordinator() || isHighest())) {
+ startElection(true); // to do, if a member disappears, only the coordinator can start
}
} catch (ChannelException x) {
- log.error(sm.getString("nonBlockingCoordinator.memberDisappeared.failed"),x);
+ log.error(sm.getString("nonBlockingCoordinator.memberDisappeared.failed"), x);
}
}
public boolean isHighest() {
Member local = getLocalMember(false);
- if ( membership.getMembers().length == 0 ) {
+ if (membership.getMembers().length == 0) {
return true;
} else {
- return AbsoluteOrder.comp.compare(local,membership.getMembers()[0])<=0;
+ return AbsoluteOrder.comp.compare(local, membership.getMembers()[0]) <= 0;
}
}
@@ -594,15 +618,16 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
public void heartbeat() {
try {
Member local = getLocalMember(false);
- if ( view != null && (Arrays.diff(view,membership,local).length != 0 || Arrays.diff(membership,view,local).length != 0) ) {
- if ( isHighest() ) {
+ if (view != null && (Arrays.diff(view, membership, local).length != 0 ||
+ Arrays.diff(membership, view, local).length != 0)) {
+ if (isHighest()) {
fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_START_ELECT, this,
sm.getString("nonBlockingCoordinator.heartbeat.inconsistency")));
startElection(true);
}
}
- } catch ( Exception x ){
- log.error(sm.getString("nonBlockingCoordinator.heartbeat.failed"),x);
+ } catch (Exception x) {
+ log.error(sm.getString("nonBlockingCoordinator.heartbeat.failed"), x);
} finally {
super.heartbeat();
}
@@ -629,26 +654,26 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
@Override
public Member getLocalMember(boolean incAlive) {
Member local = super.getLocalMember(incAlive);
- if ( view == null && (local != null)) {
+ if (view == null && (local != null)) {
setupMembership();
}
return local;
}
protected synchronized void setupMembership() {
- if ( membership == null ) {
- membership = new Membership(super.getLocalMember(true),AbsoluteOrder.comp,false);
+ if (membership == null) {
+ membership = new Membership(super.getLocalMember(true), AbsoluteOrder.comp, false);
}
}
-//============================================================================================================
-// HELPER CLASSES FOR COORDINATION
-//============================================================================================================
+ // ============================================================================================================
+ // HELPER CLASSES FOR COORDINATION
+ // ============================================================================================================
public static class CoordinationMessage {
- //X{A-ldr, A-src, mbrs-A,B,C,D}
+ // X{A-ldr, A-src, mbrs-A,B,C,D}
protected final XByteBuffer buf;
protected Member leader;
protected Member source;
@@ -661,12 +686,8 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
parse();
}
- public CoordinationMessage(Member leader,
- Member source,
- Member[] view,
- UniqueId id,
- byte[] type) {
- this.buf = new XByteBuffer(4096,false);
+ public CoordinationMessage(Member leader, Member source, Member[] view, UniqueId id, byte[] type) {
+ this.buf = new XByteBuffer(4096, false);
this.leader = leader;
this.source = source;
this.view = view;
@@ -681,35 +702,35 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
}
public Member getLeader() {
- if ( leader == null ) {
+ if (leader == null) {
parse();
}
return leader;
}
public Member getSource() {
- if ( source == null ) {
+ if (source == null) {
parse();
}
return source;
}
public UniqueId getId() {
- if ( id == null ) {
+ if (id == null) {
parse();
}
return id;
}
public Member[] getMembers() {
- if ( view == null ) {
+ if (view == null) {
parse();
}
return view;
}
public byte[] getType() {
- if (type == null ) {
+ if (type == null) {
parse();
}
return type;
@@ -720,36 +741,36 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
}
public void parse() {
- //header
+ // header
int offset = 16;
- //leader
- int ldrLen = XByteBuffer.toInt(buf.getBytesDirect(),offset);
+ // leader
+ int ldrLen = XByteBuffer.toInt(buf.getBytesDirect(), offset);
offset += 4;
byte[] ldr = new byte[ldrLen];
- System.arraycopy(buf.getBytesDirect(),offset,ldr,0,ldrLen);
+ System.arraycopy(buf.getBytesDirect(), offset, ldr, 0, ldrLen);
leader = MemberImpl.getMember(ldr);
offset += ldrLen;
- //source
- int srcLen = XByteBuffer.toInt(buf.getBytesDirect(),offset);
+ // source
+ int srcLen = XByteBuffer.toInt(buf.getBytesDirect(), offset);
offset += 4;
byte[] src = new byte[srcLen];
- System.arraycopy(buf.getBytesDirect(),offset,src,0,srcLen);
+ System.arraycopy(buf.getBytesDirect(), offset, src, 0, srcLen);
source = MemberImpl.getMember(src);
offset += srcLen;
- //view
- int mbrCount = XByteBuffer.toInt(buf.getBytesDirect(),offset);
+ // view
+ int mbrCount = XByteBuffer.toInt(buf.getBytesDirect(), offset);
offset += 4;
view = new Member[mbrCount];
- for (int i=0; i<view.length; i++ ) {
- int mbrLen = XByteBuffer.toInt(buf.getBytesDirect(),offset);
+ for (int i = 0; i < view.length; i++) {
+ int mbrLen = XByteBuffer.toInt(buf.getBytesDirect(), offset);
offset += 4;
byte[] mbr = new byte[mbrLen];
System.arraycopy(buf.getBytesDirect(), offset, mbr, 0, mbrLen);
view[i] = MemberImpl.getMember(mbr);
offset += mbrLen;
}
- //id
- this.id = new UniqueId(buf.getBytesDirect(),offset,16);
+ // id
+ this.id = new UniqueId(buf.getBytesDirect(), offset, 16);
offset += 16;
type = new byte[16];
System.arraycopy(buf.getBytesDirect(), offset, type, 0, type.length);
@@ -759,28 +780,28 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
public void write() {
buf.reset();
- //header
- buf.append(COORD_HEADER,0,COORD_HEADER.length);
- //leader
- byte[] ldr = leader.getData(false,false);
+ // header
+ buf.append(COORD_HEADER, 0, COORD_HEADER.length);
+ // leader
+ byte[] ldr = leader.getData(false, false);
buf.append(ldr.length);
- buf.append(ldr,0,ldr.length);
+ buf.append(ldr, 0, ldr.length);
ldr = null;
- //source
- byte[] src = source.getData(false,false);
+ // source
+ byte[] src = source.getData(false, false);
buf.append(src.length);
- buf.append(src,0,src.length);
+ buf.append(src, 0, src.length);
src = null;
- //view
+ // view
buf.append(view.length);
for (Member member : view) {
byte[] mbr = member.getData(false, false);
buf.append(mbr.length);
buf.append(mbr, 0, mbr.length);
}
- //id
- buf.append(id.getBytes(),0,id.getBytes().length);
- buf.append(type,0,type.length);
+ // id
+ buf.append(id.getBytes(), 0, id.getBytes().length);
+ buf.append(type, 0, type.length);
}
}
@@ -819,14 +840,15 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
final String info;
final Membership view;
final Membership suggestedView;
- public CoordinationEvent(int type,ChannelInterceptor interceptor, String info) {
+
+ public CoordinationEvent(int type, ChannelInterceptor interceptor, String info) {
this.type = type;
this.interceptor = interceptor;
- this.coord = ((NonBlockingCoordinator)interceptor).getCoordinator();
- this.mbrs = ((NonBlockingCoordinator)interceptor).membership.getMembers();
+ this.coord = ((NonBlockingCoordinator) interceptor).getCoordinator();
+ this.mbrs = ((NonBlockingCoordinator) interceptor).membership.getMembers();
this.info = info;
- this.view = ((NonBlockingCoordinator)interceptor).view;
- this.suggestedView = ((NonBlockingCoordinator)interceptor).suggestedView;
+ this.view = ((NonBlockingCoordinator) interceptor).view;
+ this.suggestedView = ((NonBlockingCoordinator) interceptor).suggestedView;
}
@Override
@@ -837,20 +859,34 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
@Override
public String getEventTypeDesc() {
switch (type) {
- case EVT_START: return "EVT_START:"+info;
- case EVT_MBR_ADD: return "EVT_MBR_ADD:"+info;
- case EVT_MBR_DEL: return "EVT_MBR_DEL:"+info;
- case EVT_START_ELECT: return "EVT_START_ELECT:"+info;
- case EVT_PROCESS_ELECT: return "EVT_PROCESS_ELECT:"+info;
- case EVT_MSG_ARRIVE: return "EVT_MSG_ARRIVE:"+info;
- case EVT_PRE_MERGE: return "EVT_PRE_MERGE:"+info;
- case EVT_POST_MERGE: return "EVT_POST_MERGE:"+info;
- case EVT_WAIT_FOR_MSG: return "EVT_WAIT_FOR_MSG:"+info;
- case EVT_SEND_MSG: return "EVT_SEND_MSG:"+info;
- case EVT_STOP: return "EVT_STOP:"+info;
- case EVT_CONF_RX: return "EVT_CONF_RX:"+info;
- case EVT_ELECT_ABANDONED: return "EVT_ELECT_ABANDONED:"+info;
- default: return "Unknown";
+ case EVT_START:
+ return "EVT_START:" + info;
+ case EVT_MBR_ADD:
+ return "EVT_MBR_ADD:" + info;
+ case EVT_MBR_DEL:
+ return "EVT_MBR_DEL:" + info;
+ case EVT_START_ELECT:
+ return "EVT_START_ELECT:" + info;
+ case EVT_PROCESS_ELECT:
+ return "EVT_PROCESS_ELECT:" + info;
+ case EVT_MSG_ARRIVE:
+ return "EVT_MSG_ARRIVE:" + info;
+ case EVT_PRE_MERGE:
+ return "EVT_PRE_MERGE:" + info;
+ case EVT_POST_MERGE:
+ return "EVT_POST_MERGE:" + info;
+ case EVT_WAIT_FOR_MSG:
+ return "EVT_WAIT_FOR_MSG:" + info;
+ case EVT_SEND_MSG:
+ return "EVT_SEND_MSG:" + info;
+ case EVT_STOP:
+ return "EVT_STOP:" + info;
+ case EVT_CONF_RX:
+ return "EVT_CONF_RX:" + info;
+ case EVT_ELECT_ABANDONED:
+ return "EVT_ELECT_ABANDONED:" + info;
+ default:
+ return "Unknown";
}
}
@@ -863,8 +899,7 @@ public class NonBlockingCoordinator extends ChannelInterceptorBase {
public String toString() {
Member local = interceptor.getLocalMember(false);
return sm.getString("nonBlockingCoordinator.report", Integer.valueOf(type),
- (local != null ? local.getName() : ""),
- (coord != null ? coord.getName() : ""),
+ (local != null ? local.getName() : ""), (coord != null ? coord.getName() : ""),
Arrays.toNameString(view != null ? view.getMembers() : null),
Arrays.toNameString(suggestedView != null ? suggestedView.getMembers() : null),
Arrays.toNameString(mbrs), info);
diff --git a/java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
index 1fbf67b8a0..b1e0668372 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
@@ -31,39 +31,39 @@ import org.apache.catalina.tribes.util.StringManager;
/**
- * The order interceptor guarantees that messages are received in the same order they were
- * sent.
- * This interceptor works best with the ack=true setting. <br>
- * There is no point in
- * using this with the replicationMode="fastasynchqueue" as this mode guarantees ordering.<BR>
- * If you are using the mode ack=false replicationMode=pooled, and have a lot of concurrent threads,
- * this interceptor can really slow you down, as many messages will be completely out of order
- * and the queue might become rather large. If this is the case, then you might want to set
- * the value OrderInterceptor.maxQueue = 25 (meaning that we will never keep more than 25 messages in our queue)
- * <br><b>Configuration Options</b><br>
- * OrderInterceptor.expire=<milliseconds> - if a message arrives out of order, how long before we act on it <b>default=3000ms</b><br>
- * OrderInterceptor.maxQueue=<max queue size> - how much can the queue grow to ensure ordering.
- * This setting is useful to avoid OutOfMemoryErrors<b>default=Integer.MAX_VALUE</b><br>
- * OrderInterceptor.forwardExpired=<boolean> - this flag tells the interceptor what to
- * do when a message has expired or the queue has grown larger than the maxQueue value.
- * true means that the message is sent up the stack to the receiver that will receive and out of order message
- * false means, forget the message and reset the message counter. <b>default=true</b>
+ * The order interceptor guarantees that messages are received in the same order they were sent. This interceptor works
+ * best with the ack=true setting. <br>
+ * There is no point in using this with the replicationMode="fastasynchqueue" as this mode guarantees ordering.<BR>
+ * If you are using the mode ack=false replicationMode=pooled, and have a lot of concurrent threads, this interceptor
+ * can really slow you down, as many messages will be completely out of order and the queue might become rather large.
+ * If this is the case, then you might want to set the value OrderInterceptor.maxQueue = 25 (meaning that we will never
+ * keep more than 25 messages in our queue) <br>
+ * <b>Configuration Options</b><br>
+ * OrderInterceptor.expire=<milliseconds> - if a message arrives out of order, how long before we act on it
+ * <b>default=3000ms</b><br>
+ * OrderInterceptor.maxQueue=<max queue size> - how much can the queue grow to ensure ordering. This setting is
+ * useful to avoid OutOfMemoryErrors<b>default=Integer.MAX_VALUE</b><br>
+ * OrderInterceptor.forwardExpired=<boolean> - this flag tells the interceptor what to do when a message has
+ * expired or the queue has grown larger than the maxQueue value. true means that the message is sent up the stack to
+ * the receiver that will receive and out of order message false means, forget the message and reset the message
+ * counter. <b>default=true</b>
*/
public class OrderInterceptor extends ChannelInterceptorBase {
protected static final StringManager sm = StringManager.getManager(OrderInterceptor.class);
- private final Map<Member, Counter> outcounter = new HashMap<>();
- private final Map<Member, Counter> incounter = new HashMap<>();
- private final Map<Member, MessageOrder> incoming = new HashMap<>();
+ private final Map<Member,Counter> outcounter = new HashMap<>();
+ private final Map<Member,Counter> incounter = new HashMap<>();
+ private final Map<Member,MessageOrder> incoming = new HashMap<>();
private long expire = 3000;
private boolean forwardExpired = true;
private int maxQueue = Integer.MAX_VALUE;
final ReentrantReadWriteLock inLock = new ReentrantReadWriteLock(true);
- final ReentrantReadWriteLock outLock= new ReentrantReadWriteLock(true);
+ final ReentrantReadWriteLock outLock = new ReentrantReadWriteLock(true);
@Override
- public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
- if ( !okToProcess(msg.getOptions()) ) {
+ public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
+ throws ChannelException {
+ if (!okToProcess(msg.getOptions())) {
super.sendMessage(destination, msg, payload);
return;
}
@@ -77,10 +77,10 @@ public class OrderInterceptor extends ChannelInterceptorBase {
} finally {
outLock.writeLock().unlock();
}
- //reduce byte copy
+ // reduce byte copy
msg.getMessage().append(nr);
try {
- getNext().sendMessage(new Member[]{member}, msg, payload);
+ getNext().sendMessage(new Member[] { member }, msg, payload);
} finally {
msg.getMessage().trim(4);
}
@@ -90,42 +90,45 @@ public class OrderInterceptor extends ChannelInterceptorBase {
}
cx.addFaultyMember(x.getFaultyMembers());
}
- }//for
- if ( cx != null ) {
+ } // for
+ if (cx != null) {
throw cx;
}
}
@Override
public void messageReceived(ChannelMessage msg) {
- if ( !okToProcess(msg.getOptions()) ) {
+ if (!okToProcess(msg.getOptions())) {
super.messageReceived(msg);
return;
}
- int msgnr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
+ int msgnr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(), msg.getMessage().getLength() - 4);
msg.getMessage().trim(4);
- MessageOrder order = new MessageOrder(msgnr,(ChannelMessage)msg.deepclone());
+ MessageOrder order = new MessageOrder(msgnr, (ChannelMessage) msg.deepclone());
inLock.writeLock().lock();
try {
- if ( processIncoming(order) ) {
- processLeftOvers(msg.getAddress(),false);
+ if (processIncoming(order)) {
+ processLeftOvers(msg.getAddress(), false);
}
} finally {
inLock.writeLock().unlock();
}
}
+
protected void processLeftOvers(Member member, boolean force) {
MessageOrder tmp = incoming.get(member);
- if ( force ) {
+ if (force) {
Counter cnt = getInCounter(member);
cnt.setCounter(Integer.MAX_VALUE);
}
- if ( tmp!= null ) {
+ if (tmp != null) {
processIncoming(tmp);
}
}
+
/**
* @param order MessageOrder
+ *
* @return boolean - true if a message expired and was processed
*/
protected boolean processIncoming(MessageOrder order) {
@@ -134,16 +137,16 @@ public class OrderInterceptor extends ChannelInterceptorBase {
Counter cnt = getInCounter(member);
MessageOrder tmp = incoming.get(member);
- if ( tmp != null ) {
- order = MessageOrder.add(tmp,order);
+ if (tmp != null) {
+ order = MessageOrder.add(tmp, order);
}
- while ( (order!=null) && (order.getMsgNr() <= cnt.getCounter()) ) {
- //we are right on target. process orders
- if ( order.getMsgNr() == cnt.getCounter() ) {
+ while ((order != null) && (order.getMsgNr() <= cnt.getCounter())) {
+ // we are right on target. process orders
+ if (order.getMsgNr() == cnt.getCounter()) {
cnt.inc();
- } else if ( order.getMsgNr() > cnt.getCounter() ) {
+ } else if (order.getMsgNr() > cnt.getCounter()) {
cnt.setCounter(order.getMsgNr());
}
super.messageReceived(order.getMessage());
@@ -153,22 +156,22 @@ public class OrderInterceptor extends ChannelInterceptorBase {
MessageOrder head = order;
MessageOrder prev = null;
tmp = order;
- //flag to empty out the queue when it larger than maxQueue
- boolean empty = order!=null?order.getCount()>=maxQueue:false;
- while ( tmp != null ) {
- //process expired messages or empty out the queue
- if ( tmp.isExpired(expire) || empty ) {
- //reset the head
- if ( tmp == head ) {
+ // flag to empty out the queue when it larger than maxQueue
+ boolean empty = order != null ? order.getCount() >= maxQueue : false;
+ while (tmp != null) {
+ // process expired messages or empty out the queue
+ if (tmp.isExpired(expire) || empty) {
+ // reset the head
+ if (tmp == head) {
head = tmp.next;
}
- cnt.setCounter(tmp.getMsgNr()+1);
- if ( getForwardExpired() ) {
+ cnt.setCounter(tmp.getMsgNr() + 1);
+ if (getForwardExpired()) {
super.messageReceived(tmp.getMessage());
}
tmp.setMessage(null);
tmp = tmp.next;
- if ( prev != null ) {
+ if (prev != null) {
prev.next = tmp;
}
result = true;
@@ -177,7 +180,7 @@ public class OrderInterceptor extends ChannelInterceptorBase {
tmp = tmp.next;
}
}
- if ( head == null ) {
+ if (head == null) {
incoming.remove(member);
} else {
incoming.put(member, head);
@@ -187,18 +190,18 @@ public class OrderInterceptor extends ChannelInterceptorBase {
@Override
public void memberAdded(Member member) {
- //notify upwards
+ // notify upwards
super.memberAdded(member);
}
@Override
public void memberDisappeared(Member member) {
- //reset counters - lock free
+ // reset counters - lock free
incounter.remove(member);
outcounter.remove(member);
- //clear the remaining queue
- processLeftOvers(member,true);
- //notify upwards
+ // clear the remaining queue
+ processLeftOvers(member, true);
+ // notify upwards
super.memberDisappeared(member);
}
@@ -209,19 +212,19 @@ public class OrderInterceptor extends ChannelInterceptorBase {
protected Counter getInCounter(Member mbr) {
Counter cnt = incounter.get(mbr);
- if ( cnt == null ) {
+ if (cnt == null) {
cnt = new Counter();
- cnt.inc(); //always start at 1 for incoming
- incounter.put(mbr,cnt);
+ cnt.inc(); // always start at 1 for incoming
+ incounter.put(mbr, cnt);
}
return cnt;
}
protected Counter getOutCounter(Member mbr) {
Counter cnt = outcounter.get(mbr);
- if ( cnt == null ) {
+ if (cnt == null) {
cnt = new Counter();
- outcounter.put(mbr,cnt);
+ outcounter.put(mbr, cnt);
}
return cnt;
}
@@ -247,13 +250,14 @@ public class OrderInterceptor extends ChannelInterceptorBase {
private MessageOrder next;
private final int msgNr;
private ChannelMessage msg = null;
- public MessageOrder(int msgNr,ChannelMessage msg) {
+
+ public MessageOrder(int msgNr, ChannelMessage msg) {
this.msgNr = msgNr;
this.msg = msg;
}
public boolean isExpired(long expireTime) {
- return (System.currentTimeMillis()-received) > expireTime;
+ return (System.currentTimeMillis() - received) > expireTime;
}
public ChannelMessage getMessage() {
@@ -267,6 +271,7 @@ public class OrderInterceptor extends ChannelInterceptorBase {
public void setNext(MessageOrder order) {
this.next = order;
}
+
public MessageOrder getNext() {
return next;
}
@@ -274,7 +279,7 @@ public class OrderInterceptor extends ChannelInterceptorBase {
public int getCount() {
int counter = 1;
MessageOrder tmp = next;
- while ( tmp != null ) {
+ while (tmp != null) {
counter++;
tmp = tmp.next;
}
@@ -283,33 +288,33 @@ public class OrderInterceptor extends ChannelInterceptorBase {
@SuppressWarnings("null") // prev cannot be null
public static MessageOrder add(MessageOrder head, MessageOrder add) {
- if ( head == null ) {
+ if (head == null) {
return add;
}
- if ( add == null ) {
+ if (add == null) {
return head;
}
- if ( head == add ) {
+ if (head == add) {
return add;
}
- if ( head.getMsgNr() > add.getMsgNr() ) {
+ if (head.getMsgNr() > add.getMsgNr()) {
add.next = head;
return add;
}
MessageOrder iter = head;
MessageOrder prev = null;
- while ( iter.getMsgNr() < add.getMsgNr() && (iter.next !=null ) ) {
+ while (iter.getMsgNr() < add.getMsgNr() && (iter.next != null)) {
prev = iter;
iter = iter.next;
}
- if ( iter.getMsgNr() < add.getMsgNr() ) {
- //add after
+ if (iter.getMsgNr() < add.getMsgNr()) {
+ // add after
add.next = iter.next;
iter.next = add;
} else if (iter.getMsgNr() > add.getMsgNr()) {
- //add before
+ // add before
prev.next = add; // prev cannot be null here, warning suppressed
add.next = iter;
diff --git a/java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java b/java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java
index ebdd6aecb5..a3082c8dcb 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java
@@ -29,7 +29,6 @@ import org.apache.catalina.tribes.group.ChannelInterceptorBase;
* A dinky coordinator, just uses a sorted version of the member array.
*
* @author rnewson
- *
*/
public class SimpleCoordinator extends ChannelInterceptorBase {
@@ -86,7 +85,7 @@ public class SimpleCoordinator extends ChannelInterceptorBase {
}
final Member[] members = getMembers();
- final Member[] view = new Member[members.length+1];
+ final Member[] view = new Member[members.length + 1];
System.arraycopy(members, 0, view, 0, members.length);
view[members.length] = getLocalMember(false);
Arrays.sort(view, AbsoluteOrder.comp);
@@ -111,8 +110,7 @@ public class SimpleCoordinator extends ChannelInterceptorBase {
}
public boolean isCoordinator() {
- return view == null ? false : getLocalMember(false).equals(
- getCoordinator());
+ return view == null ? false : getLocalMember(false).equals(getCoordinator());
}
}
diff --git a/java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptor.java
index e4fa3545f5..604efca1b8 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptor.java
@@ -33,20 +33,16 @@ import org.apache.catalina.tribes.util.StringManager;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
-public class StaticMembershipInterceptor extends ChannelInterceptorBase
- implements StaticMembershipInterceptorMBean {
+public class StaticMembershipInterceptor extends ChannelInterceptorBase implements StaticMembershipInterceptorMBean {
private static final Log log = LogFactory.getLog(StaticMembershipInterceptor.class);
- protected static final StringManager sm =
- StringManager.getManager(StaticMembershipInterceptor.class);
+ protected static final StringManager sm = StringManager.getManager(StaticMembershipInterceptor.class);
- protected static final byte[] MEMBER_START = new byte[] {
- 76, 111, 99, 97, 108, 32, 83, 116, 97, 116, 105, 99, 77, 101, 109, 98, 101, 114, 32, 78,
- 111, 116, 105, 102, 105, 99, 97, 116, 105, 111, 110, 32, 68, 97, 116, 97};
+ protected static final byte[] MEMBER_START = new byte[] { 76, 111, 99, 97, 108, 32, 83, 116, 97, 116, 105, 99, 77,
+ 101, 109, 98, 101, 114, 32, 78, 111, 116, 105, 102, 105, 99, 97, 116, 105, 111, 110, 32, 68, 97, 116, 97 };
- protected static final byte[] MEMBER_STOP = new byte[] {
- 76, 111, 99, 97, 108, 32, 83, 116, 97, 116, 105, 99, 77, 101, 109, 98, 101, 114, 32, 83,
- 104, 117, 116, 100, 111, 119, 110, 32, 68, 97, 116, 97};
+ protected static final byte[] MEMBER_STOP = new byte[] { 76, 111, 99, 97, 108, 32, 83, 116, 97, 116, 105, 99, 77,
+ 101, 109, 98, 101, 114, 32, 83, 104, 117, 116, 100, 111, 119, 110, 32, 68, 97, 116, 97 };
protected final ArrayList<Member> members = new ArrayList<>();
protected Member localMember = null;
@@ -86,7 +82,7 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
super.memberAdded(member);
}
} else if (msg.getMessage().getLength() == MEMBER_STOP.length &&
- Arrays.equals(MEMBER_STOP, msg.getMessage().getBytes())) {
+ Arrays.equals(MEMBER_STOP, msg.getMessage().getBytes())) {
// receive member shutdown
Member member = getMember(msg.getAddress());
if (member != null) {
@@ -104,12 +100,12 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
@Override
public boolean hasMembers() {
- return super.hasMembers() || (members.size()>0);
+ return super.hasMembers() || (members.size() > 0);
}
@Override
public Member[] getMembers() {
- if ( members.size() == 0 ) {
+ if (members.size() == 0) {
return super.getMembers();
} else {
synchronized (members) {
@@ -123,13 +119,13 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
}
AbsoluteOrder.absoluteOrder(result);
return result;
- }//sync
- }//end if
+ } // sync
+ } // end if
}
@Override
public Member getMember(Member mbr) {
- if ( members.contains(mbr) ) {
+ if (members.contains(mbr)) {
return members.get(members.indexOf(mbr));
} else {
return super.getMember(mbr);
@@ -138,7 +134,7 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
@Override
public Member getLocalMember(boolean incAlive) {
- if (this.localMember != null ) {
+ if (this.localMember != null) {
return localMember;
} else {
return super.getLocalMember(incAlive);
@@ -152,10 +148,10 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
*/
@Override
public void start(int svc) throws ChannelException {
- if ( (Channel.SND_RX_SEQ&svc)==Channel.SND_RX_SEQ ) {
+ if ((Channel.SND_RX_SEQ & svc) == Channel.SND_RX_SEQ) {
super.start(Channel.SND_RX_SEQ);
}
- if ( (Channel.SND_TX_SEQ&svc)==Channel.SND_TX_SEQ ) {
+ if ((Channel.SND_TX_SEQ & svc) == Channel.SND_TX_SEQ) {
super.start(Channel.SND_TX_SEQ);
}
final ChannelInterceptorBase base = this;
@@ -164,7 +160,7 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
Runnable r = () -> {
base.memberAdded(member);
if (getfirstInterceptor().getMember(member) != null) {
- sendLocalMember(new Member[]{member});
+ sendLocalMember(new Member[] { member });
}
};
executor.execute(r);
@@ -176,7 +172,7 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
TcpPingInterceptor pingInterceptor = null;
ChannelInterceptor prev = getPrevious();
while (prev != null) {
- if (prev instanceof TcpFailureDetector ) {
+ if (prev instanceof TcpFailureDetector) {
failureDetector = (TcpFailureDetector) prev;
}
if (prev instanceof TcpPingInterceptor) {
@@ -209,7 +205,7 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
try {
sendMemberMessage(members, MEMBER_START);
} catch (ChannelException cx) {
- log.warn(sm.getString("staticMembershipInterceptor.sendLocalMember.failed"),cx);
+ log.warn(sm.getString("staticMembershipInterceptor.sendLocalMember.failed"), cx);
}
}
@@ -217,7 +213,7 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
try {
sendMemberMessage(members, MEMBER_STOP);
} catch (ChannelException cx) {
- log.warn(sm.getString("staticMembershipInterceptor.sendShutdown.failed"),cx);
+ log.warn(sm.getString("staticMembershipInterceptor.sendShutdown.failed"), cx);
}
}
@@ -232,7 +228,7 @@ public class StaticMembershipInterceptor extends ChannelInterceptorBase
}
protected void sendMemberMessage(Member[] members, byte[] message) throws ChannelException {
- if ( members == null || members.length == 0 ) {
+ if (members == null || members.length == 0) {
return;
}
ChannelData data = new ChannelData(true);
diff --git a/java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptorMBean.java b/java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptorMBean.java
index b090237a11..7c5d7c1792 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptorMBean.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptorMBean.java
@@ -18,7 +18,7 @@ package org.apache.catalina.tribes.group.interceptors;
import org.apache.catalina.tribes.Member;
-public interface StaticMembershipInterceptorMBean {
+public interface StaticMembershipInterceptorMBean {
int getOptionFlag();
diff --git a/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java b/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java
index dbfd3acd37..acc6422062 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java
@@ -42,13 +42,11 @@ import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
/**
- * The TcpFailureDetector is a useful interceptor that adds reliability to the
- * membership layer.
+ * The TcpFailureDetector is a useful interceptor that adds reliability to the membership layer.
* <p>
- * If the network is busy, or the system is busy so that the membership receiver thread
- * is not getting enough time to update its table, members can be "timed out"
- * This failure detector will intercept the memberDisappeared message(unless its a true shutdown message)
- * and connect to the member using TCP.
+ * If the network is busy, or the system is busy so that the membership receiver thread is not getting enough time to
+ * update its table, members can be "timed out" This failure detector will intercept the memberDisappeared
+ * message(unless its a true shutdown message) and connect to the member using TCP.
* <p>
* The TcpFailureDetector works in two ways:
* <ol>
@@ -61,76 +59,76 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
private static final Log log = LogFactory.getLog(TcpFailureDetector.class);
protected static final StringManager sm = StringManager.getManager(TcpFailureDetector.class);
- protected static final byte[] TCP_FAIL_DETECT = new byte[] {
- 79, -89, 115, 72, 121, -126, 67, -55, -97, 111, -119, -128, -95, 91, 7, 20,
- 125, -39, 82, 91, -21, -15, 67, -102, -73, 126, -66, -113, -127, 103, 30, -74,
- 55, 21, -66, -121, 69, 126, 76, -88, -65, 10, 77, 19, 83, 56, 21, 50,
- 85, -10, -108, -73, 58, -6, 64, 120, -111, 4, 125, -41, 114, -124, -64, -43};
+ protected static final byte[] TCP_FAIL_DETECT = new byte[] { 79, -89, 115, 72, 121, -126, 67, -55, -97, 111, -119,
+ -128, -95, 91, 7, 20, 125, -39, 82, 91, -21, -15, 67, -102, -73, 126, -66, -113, -127, 103, 30, -74, 55, 21,
+ -66, -121, 69, 126, 76, -88, -65, 10, 77, 19, 83, 56, 21, 50, 85, -10, -108, -73, 58, -6, 64, 120, -111, 4,
+ 125, -41, 114, -124, -64, -43 };
- protected long connectTimeout = 1000;//1 second default
+ protected long connectTimeout = 1000;// 1 second default
protected boolean performSendTest = true;
protected boolean performReadTest = false;
- protected long readTestTimeout = 5000;//5 seconds
+ protected long readTestTimeout = 5000;// 5 seconds
protected Membership membership = null;
- protected final HashMap<Member, Long> removeSuspects = new HashMap<>();
+ protected final HashMap<Member,Long> removeSuspects = new HashMap<>();
- protected final HashMap<Member, Long> addSuspects = new HashMap<>();
+ protected final HashMap<Member,Long> addSuspects = new HashMap<>();
protected int removeSuspectsTimeout = 300; // 5 minutes
@Override
- public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
+ public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
+ throws ChannelException {
try {
super.sendMessage(destination, msg, payload);
- }catch ( ChannelException cx ) {
+ } catch (ChannelException cx) {
FaultyMember[] mbrs = cx.getFaultyMembers();
for (FaultyMember mbr : mbrs) {
- if (mbr.getCause() != null &&
- (!(mbr.getCause() instanceof RemoteProcessException))) {//RemoteProcessException's are ok
+ if (mbr.getCause() != null && (!(mbr.getCause() instanceof RemoteProcessException))) {// RemoteProcessException's
+ // are ok
this.memberDisappeared(mbr.getMember());
- }//end if
- }//for
+ } // end if
+ } // for
throw cx;
}
}
@Override
public void messageReceived(ChannelMessage msg) {
- //catch incoming
+ // catch incoming
boolean process = true;
- if ( okToProcess(msg.getOptions()) ) {
- //check to see if it is a testMessage, if so, process = false
- process = ( (msg.getMessage().getLength() != TCP_FAIL_DETECT.length) ||
- (!Arrays.equals(TCP_FAIL_DETECT,msg.getMessage().getBytes()) ) );
- }//end if
-
- //ignore the message, it doesn't have the flag set
- if ( process ) {
+ if (okToProcess(msg.getOptions())) {
+ // check to see if it is a testMessage, if so, process = false
+ process = ((msg.getMessage().getLength() != TCP_FAIL_DETECT.length) ||
+ (!Arrays.equals(TCP_FAIL_DETECT, msg.getMessage().getBytes())));
+ } // end if
+
+ // ignore the message, it doesn't have the flag set
+ if (process) {
super.messageReceived(msg);
- } else if ( log.isDebugEnabled() ) {
+ } else if (log.isDebugEnabled()) {
log.debug(sm.getString("tcpFailureDetector.recievedPacket", msg));
}
- }//messageReceived
+ }// messageReceived
@Override
public void memberAdded(Member member) {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
boolean notify = false;
synchronized (membership) {
if (removeSuspects.containsKey(member)) {
- //previously marked suspect, system below picked up the member again
+ // previously marked suspect, system below picked up the member again
removeSuspects.remove(member);
- } else if (membership.getMember(member) == null){
- //if we add it here, then add it upwards too
- //check to see if it is alive
+ } else if (membership.getMember(member) == null) {
+ // if we add it here, then add it upwards too
+ // check to see if it is alive
if (memberAlive(member)) {
membership.memberAlive(member);
addSuspects.remove(member);
@@ -142,17 +140,17 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
}
}
}
- if ( notify ) {
+ if (notify) {
super.memberAdded(member);
}
}
@Override
public void memberDisappeared(Member member) {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
- boolean shutdown = Arrays.equals(member.getCommand(),Member.SHUTDOWN_PAYLOAD);
+ boolean shutdown = Arrays.equals(member.getCommand(), Member.SHUTDOWN_PAYLOAD);
if (shutdown) {
synchronized (membership) {
if (!membership.contains(member)) {
@@ -167,19 +165,19 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
super.memberDisappeared(member);
} else {
boolean notify = false;
- if(log.isInfoEnabled()) {
+ if (log.isInfoEnabled()) {
log.info(sm.getString("tcpFailureDetector.memberDisappeared.verify", member));
}
synchronized (membership) {
if (!membership.contains(member)) {
- if(log.isInfoEnabled()) {
+ if (log.isInfoEnabled()) {
log.info(sm.getString("tcpFailureDetector.already.disappeared", member));
}
return;
}
- //check to see if the member really is gone
+ // check to see if the member really is gone
if (!memberAlive(member)) {
- //not correct, we need to maintain the map
+ // not correct, we need to maintain the map
membership.removeMember(member);
removeSuspects.remove(member);
if (member instanceof StaticMember) {
@@ -187,17 +185,17 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
}
notify = true;
} else {
- //add the member as suspect
+ // add the member as suspect
removeSuspects.put(member, Long.valueOf(System.currentTimeMillis()));
}
}
- if ( notify ) {
- if(log.isInfoEnabled()) {
+ if (notify) {
+ if (log.isInfoEnabled()) {
log.info(sm.getString("tcpFailureDetector.member.disappeared", member));
}
super.memberDisappeared(member);
} else {
- if(log.isInfoEnabled()) {
+ if (log.isInfoEnabled()) {
log.info(sm.getString("tcpFailureDetector.still.alive", member));
}
}
@@ -206,7 +204,7 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
@Override
public boolean hasMembers() {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
return membership.hasMembers();
@@ -214,7 +212,7 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
@Override
public Member[] getMembers() {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
return membership.getMembers();
@@ -222,7 +220,7 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
@Override
public Member getMember(Member mbr) {
- if ( membership == null ) {
+ if (membership == null) {
setupMembership();
}
return membership.getMember(mbr);
@@ -253,12 +251,12 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
}
}
} catch (Exception x) {
- log.warn(sm.getString("tcpFailureDetector.heartbeat.failed"),x);
+ log.warn(sm.getString("tcpFailureDetector.heartbeat.failed"), x);
}
}
protected void performForcedCheck() {
- //update all alive times
+ // update all alive times
Member[] members = super.getMembers();
for (int i = 0; members != null && i < members.length; i++) {
if (memberAlive(members[i])) {
@@ -267,7 +265,7 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
}
addSuspects.remove(members[i]);
} else {
- if (membership.getMember(members[i])!=null) {
+ if (membership.getMember(members[i]) != null) {
membership.removeMember(members[i]);
removeSuspects.remove(members[i]);
if (members[i] instanceof StaticMember) {
@@ -275,13 +273,13 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
}
super.memberDisappeared(members[i]);
}
- } //end if
- } //for
+ } // end if
+ } // for
}
protected void performBasicCheck() {
- //update all alive times
+ // update all alive times
Member[] members = super.getMembers();
for (int i = 0; members != null && i < members.length; i++) {
if (addSuspects.containsKey(members[i]) && membership.getMember(members[i]) == null) {
@@ -289,18 +287,18 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
continue;
}
if (membership.memberAlive(members[i])) {
- //we don't have this one in our membership, check to see if the member is alive
+ // we don't have this one in our membership, check to see if the member is alive
if (memberAlive(members[i])) {
log.warn(sm.getString("tcpFailureDetector.performBasicCheck.memberAdded", members[i]));
super.memberAdded(members[i]);
} else {
membership.removeMember(members[i]);
- } //end if
- } //end if
- } //for
+ } // end if
+ } // end if
+ } // for
- //check suspect members if they are still alive,
- //if not, simply issue the memberDisappeared message
+ // check suspect members if they are still alive,
+ // if not, simply issue the memberDisappeared message
Member[] keys = removeSuspects.keySet().toArray(new Member[0]);
for (Member m : keys) {
if (membership.getMember(m) != null && (!memberAlive(m))) {
@@ -324,8 +322,8 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
}
}
- //check add suspects members if they are alive now,
- //if they are, simply issue the memberAdded message
+ // check add suspects members if they are alive now,
+ // if they are, simply issue the memberAdded message
keys = addSuspects.keySet().toArray(new Member[0]);
for (Member m : keys) {
if (membership.getMember(m) == null && (memberAlive(m))) {
@@ -335,42 +333,41 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
if (log.isInfoEnabled()) {
log.info(sm.getString("tcpFailureDetector.suspectMember.alive", m));
}
- } //end if
+ } // end if
}
}
protected synchronized void setupMembership() {
- if ( membership == null ) {
+ if (membership == null) {
membership = new Membership(super.getLocalMember(true));
}
}
protected boolean memberAlive(Member mbr) {
- return memberAlive(mbr,TCP_FAIL_DETECT,performSendTest,performReadTest,readTestTimeout,connectTimeout,getOptionFlag());
+ return memberAlive(mbr, TCP_FAIL_DETECT, performSendTest, performReadTest, readTestTimeout, connectTimeout,
+ getOptionFlag());
}
- protected boolean memberAlive(Member mbr, byte[] msgData,
- boolean sendTest, boolean readTest,
- long readTimeout, long conTimeout,
- int optionFlag) {
- //could be a shutdown notification
- if ( Arrays.equals(mbr.getCommand(),Member.SHUTDOWN_PAYLOAD) ) {
+ protected boolean memberAlive(Member mbr, byte[] msgData, boolean sendTest, boolean readTest, long readTimeout,
+ long conTimeout, int optionFlag) {
+ // could be a shutdown notification
+ if (Arrays.equals(mbr.getCommand(), Member.SHUTDOWN_PAYLOAD)) {
return false;
}
try (Socket socket = new Socket()) {
InetAddress ia = InetAddress.getByAddress(mbr.getHost());
InetSocketAddress addr = new InetSocketAddress(ia, mbr.getPort());
- socket.setSoTimeout((int)readTimeout);
+ socket.setSoTimeout((int) readTimeout);
socket.connect(addr, (int) conTimeout);
- if ( sendTest ) {
+ if (sendTest) {
ChannelData data = new ChannelData(true);
data.setAddress(getLocalMember(false));
- data.setMessage(new XByteBuffer(msgData,false));
+ data.setMessage(new XByteBuffer(msgData, false));
data.setTimestamp(System.currentTimeMillis());
int options = optionFlag | Channel.SEND_OPTIONS_BYTE_MESSAGE;
- if ( readTest ) {
+ if (readTest) {
options = (options | Channel.SEND_OPTIONS_USE_ACK);
} else {
options = (options & (~Channel.SEND_OPTIONS_USE_ACK));
@@ -378,16 +375,16 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai
data.setOptions(options);
byte[] message = XByteBuffer.createDataPackage(data);
socket.getOutputStream().write(message);
- if ( readTest ) {
+ if (readTest) {
int length = socket.getInputStream().read(message);
return length > 0;
}
- }//end if
+ } // end if
return true;
} catch (SocketTimeoutException | ConnectException | NoRouteToHostException noop) {
- //do nothing, we couldn't connect
+ // do nothing, we couldn't connect
} catch (Exception x) {
- log.error(sm.getString("tcpFailureDetector.failureDetection.failed", mbr),x);
+ log.error(sm.getString("tcpFailureDetector.failureDetection.failed", mbr), x);
}
return false;
}
diff --git a/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java
index 06a21c5540..d340c9b955 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java
@@ -32,9 +32,8 @@ import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
/**
- * Sends a ping to all members.
- * Configure this interceptor with the TcpFailureDetector below it,
- * and the TcpFailureDetector will act as the membership guide.
+ * Sends a ping to all members. Configure this interceptor with the TcpFailureDetector below it, and the
+ * TcpFailureDetector will act as the membership guide.
*/
public class TcpPingInterceptor extends ChannelInterceptorBase implements TcpPingInterceptorMBean {
@@ -42,13 +41,12 @@ public class TcpPingInterceptor extends ChannelInterceptorBase implements TcpPin
private static final Log log = LogFactory.getLog(TcpPingInterceptor.class);
protected static final StringManager sm = StringManager.getManager(TcpPingInterceptor.class);
- protected static final byte[] TCP_PING_DATA = new byte[] {
- 79, -89, 115, 72, 121, -33, 67, -55, -97, 111, -119, -128, -95, 91, 7, 20,
- 125, -39, 82, 91, -21, -33, 67, -102, -73, 126, -66, -113, -127, 103, 30, -74,
- 55, 21, -66, -121, 69, 33, 76, -88, -65, 10, 77, 19, 83, 56, 21, 50,
- 85, -10, -108, -73, 58, -33, 33, 120, -111, 4, 125, -41, 114, -124, -64, -43};
+ protected static final byte[] TCP_PING_DATA = new byte[] { 79, -89, 115, 72, 121, -33, 67, -55, -97, 111, -119,
+ -128, -95, 91, 7, 20, 125, -39, 82, 91, -21, -33, 67, -102, -73, 126, -66, -113, -127, 103, 30, -74, 55, 21,
+ -66, -121, 69, 33, 76, -88, -65, 10, 77, 19, 83, 56, 21, 50, 85, -10, -108, -73, 58, -33, 33, 120, -111, 4,
+ 125, -41, 114, -124, -64, -43 };
- protected long interval = 1000; //1 second
+ protected long interval = 1000; // 1 second
protected boolean useThread = false;
protected boolean staticOnly = false;
@@ -63,25 +61,25 @@ public class TcpPingInterceptor extends ChannelInterceptorBase implements TcpPin
public synchronized void start(int svc) throws ChannelException {
super.start(svc);
running = true;
- if ( thread == null && useThread) {
+ if (thread == null && useThread) {
thread = new PingThread();
thread.setDaemon(true);
String channelName = "";
if (getChannel().getName() != null) {
channelName = "[" + getChannel().getName() + "]";
}
- thread.setName("TcpPingInterceptor.PingThread" + channelName +"-"+cnt.addAndGet(1));
+ thread.setName("TcpPingInterceptor.PingThread" + channelName + "-" + cnt.addAndGet(1));
thread.start();
}
- //acquire the interceptors to invoke on send ping events
+ // acquire the interceptors to invoke on send ping events
ChannelInterceptor next = getNext();
- while ( next != null ) {
- if ( next instanceof TcpFailureDetector ) {
- failureDetector = new WeakReference<>((TcpFailureDetector)next);
+ while (next != null) {
+ if (next instanceof TcpFailureDetector) {
+ failureDetector = new WeakReference<>((TcpFailureDetector) next);
}
- if ( next instanceof StaticMembershipInterceptor ) {
- staticMembers = new WeakReference<>((StaticMembershipInterceptor)next);
+ if (next instanceof StaticMembershipInterceptor) {
+ staticMembers = new WeakReference<>((StaticMembershipInterceptor) next);
}
next = next.getNext();
}
@@ -133,15 +131,13 @@ public class TcpPingInterceptor extends ChannelInterceptorBase implements TcpPin
}
protected void sendPing() {
- TcpFailureDetector tcpFailureDetector =
- failureDetector != null ? failureDetector.get() : null;
+ TcpFailureDetector tcpFailureDetector = failureDetector != null ? failureDetector.get() : null;
if (tcpFailureDetector != null) {
// We have a reference to the failure detector
// Piggy back on it
tcpFailureDetector.checkMembers(true);
} else {
- StaticMembershipInterceptor smi =
- staticOnly && staticMembers != null ? staticMembers.get() : null;
+ StaticMembershipInterceptor smi = staticOnly && staticMembers != null ? staticMembers.get() : null;
if (smi != null) {
sendPingMessage(smi.getMembers());
} else {
@@ -151,38 +147,38 @@ public class TcpPingInterceptor extends ChannelInterceptorBase implements TcpPin
}
protected void sendPingMessage(Member[] members) {
- if ( members == null || members.length == 0 ) {
+ if (members == null || members.length == 0) {
return;
}
- ChannelData data = new ChannelData(true);//generates a unique Id
+ ChannelData data = new ChannelData(true);// generates a unique Id
data.setAddress(getLocalMember(false));
data.setTimestamp(System.currentTimeMillis());
data.setOptions(getOptionFlag());
data.setMessage(new XByteBuffer(TCP_PING_DATA, false));
try {
super.sendMessage(members, data, null);
- }catch (ChannelException x) {
- log.warn(sm.getString("tcpPingInterceptor.ping.failed"),x);
+ } catch (ChannelException x) {
+ log.warn(sm.getString("tcpPingInterceptor.ping.failed"), x);
}
}
@Override
public void messageReceived(ChannelMessage msg) {
- //catch incoming
+ // catch incoming
boolean process = true;
- if ( okToProcess(msg.getOptions()) ) {
- //check to see if it is a ping message, if so, process = false
- process = ( (msg.getMessage().getLength() != TCP_PING_DATA.length) ||
- (!Arrays.equals(TCP_PING_DATA,msg.getMessage().getBytes()) ) );
- }//end if
-
- //ignore the message, it doesn't have the flag set
- if ( process ) {
+ if (okToProcess(msg.getOptions())) {
+ // check to see if it is a ping message, if so, process = false
+ process = ((msg.getMessage().getLength() != TCP_PING_DATA.length) ||
+ (!Arrays.equals(TCP_PING_DATA, msg.getMessage().getBytes())));
+ } // end if
+
+ // ignore the message, it doesn't have the flag set
+ if (process) {
super.messageReceived(msg);
- } else if ( log.isTraceEnabled() ) {
+ } else if (log.isTraceEnabled()) {
log.trace("Received a TCP ping packet:" + msg);
}
- }//messageReceived
+ }// messageReceived
protected class PingThread extends Thread {
@Override
@@ -191,12 +187,12 @@ public class TcpPingInterceptor extends ChannelInterceptorBase implements TcpPin
try {
sleep(interval);
sendPing();
- }catch ( InterruptedException ix ) {
+ } catch (InterruptedException ix) {
// Ignore. Probably triggered by a call to stop().
// In the highly unlikely event it was a different trigger,
// simply ignore it and continue.
- }catch ( Exception x ) {
- log.warn(sm.getString("tcpPingInterceptor.pingFailed.pingThread"),x);
+ } catch (Exception x) {
+ log.warn(sm.getString("tcpPingInterceptor.pingFailed.pingThread"), x);
}
}
}
diff --git a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
index 05410b8a2f..d8552e9db0 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java
@@ -31,8 +31,7 @@ import org.apache.catalina.tribes.util.StringManager;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
-public class ThroughputInterceptor extends ChannelInterceptorBase
- implements ThroughputInterceptorMBean {
+public class ThroughputInterceptor extends ChannelInterceptorBase implements ThroughputInterceptorMBean {
private static final Log log = LogFactory.getLog(ThroughputInterceptor.class);
protected static final StringManager sm = StringManager.getManager(ThroughputInterceptor.class);
@@ -53,23 +52,24 @@ public class ThroughputInterceptor extends ChannelInterceptorBase
@Override
- public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
- if ( access.addAndGet(1) == 1 ) {
+ public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
+ throws ChannelException {
+ if (access.addAndGet(1) == 1) {
txStart = System.currentTimeMillis();
}
- long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
+ long bytes = XByteBuffer.getDataPackageLength(((ChannelData) msg).getDataPackageLength());
try {
super.sendMessage(destination, msg, payload);
- }catch ( ChannelException x ) {
+ } catch (ChannelException x) {
msgTxErr.addAndGet(1);
- if ( access.get() == 1 ) {
+ if (access.get() == 1) {
access.addAndGet(-1);
}
throw x;
}
- mbTx += (bytes*destination.length)/(1024d*1024d);
- mbAppTx += bytes/(1024d*1024d);
- if ( access.addAndGet(-1) == 0 ) {
+ mbTx += (bytes * destination.length) / (1024d * 1024d);
+ mbAppTx += bytes / (1024d * 1024d);
+ if (access.addAndGet(-1) == 0) {
long stop = System.currentTimeMillis();
timeTx += (stop - txStart) / 1000d;
if ((msgTxCnt.get() / (double) interval) >= lastCnt) {
@@ -82,13 +82,13 @@ public class ThroughputInterceptor extends ChannelInterceptorBase
@Override
public void messageReceived(ChannelMessage msg) {
- if ( rxStart == 0 ) {
+ if (rxStart == 0) {
rxStart = System.currentTimeMillis();
}
- long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
- mbRx += bytes/(1024d*1024d);
+ long bytes = XByteBuffer.getDataPackageLength(((ChannelData) msg).getDataPackageLength());
+ mbRx += bytes / (1024d * 1024d);
msgRxCnt.addAndGet(1);
- if ( msgRxCnt.get() % interval == 0 ) {
+ if (msgRxCnt.get() % interval == 0) {
report(timeTx);
}
super.messageReceived(msg);
@@ -97,12 +97,10 @@ public class ThroughputInterceptor extends ChannelInterceptorBase
@Override
public void report(double timeTx) {
- if ( log.isInfoEnabled() ) {
- log.info(sm.getString("throughputInterceptor.report",
- msgTxCnt, df.format(mbTx), df.format(mbAppTx), df.format(timeTx),
- df.format(mbTx/timeTx), df.format(mbAppTx/timeTx), msgTxErr, msgRxCnt,
- df.format(mbRx/((System.currentTimeMillis()-rxStart)/(double)1000)),
- df.format(mbRx)));
+ if (log.isInfoEnabled()) {
+ log.info(sm.getString("throughputInterceptor.report", msgTxCnt, df.format(mbTx), df.format(mbAppTx),
+ df.format(timeTx), df.format(mbTx / timeTx), df.format(mbAppTx / timeTx), msgTxErr, msgRxCnt,
+ df.format(mbRx / ((System.currentTimeMillis() - rxStart) / (double) 1000)), df.format(mbRx)));
}
}
diff --git a/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java
index 0b463fe197..a55e0cd7ef 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java
@@ -33,40 +33,42 @@ import org.apache.juli.logging.LogFactory;
public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase {
- private static final byte[] START_DATA = new byte[] {113, 1, -58, 2, -34, -60, 75, -78, -101, -12, 32, -29, 32, 111, -40, 4};
- private static final byte[] END_DATA = new byte[] {54, -13, 90, 110, 47, -31, 75, -24, -81, -29, 36, 52, -58, 77, -110, 56};
+ private static final byte[] START_DATA =
+ new byte[] { 113, 1, -58, 2, -34, -60, 75, -78, -101, -12, 32, -29, 32, 111, -40, 4 };
+ private static final byte[] END_DATA =
+ new byte[] { 54, -13, 90, 110, 47, -31, 75, -24, -81, -29, 36, 52, -58, 77, -110, 56 };
private static final Log log = LogFactory.getLog(TwoPhaseCommitInterceptor.class);
protected static final StringManager sm = StringManager.getManager(TwoPhaseCommitInterceptor.class);
- protected final HashMap<UniqueId, MapEntry> messages = new HashMap<>();
- protected long expire = 1000 * 60; //one minute expiration
+ protected final HashMap<UniqueId,MapEntry> messages = new HashMap<>();
+ protected long expire = 1000 * 60; // one minute expiration
protected boolean deepclone = true;
@Override
- public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws
- ChannelException {
- //todo, optimize, if destination.length==1, then we can do
- //msg.setOptions(msg.getOptions() & (~getOptionFlag())
- //and just send one message
- if (okToProcess(msg.getOptions()) ) {
+ public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
+ throws ChannelException {
+ // todo, optimize, if destination.length==1, then we can do
+ // msg.setOptions(msg.getOptions() & (~getOptionFlag())
+ // and just send one message
+ if (okToProcess(msg.getOptions())) {
super.sendMessage(destination, msg, null);
ChannelMessage confirmation = null;
- if ( deepclone ) {
- confirmation = (ChannelMessage)msg.deepclone();
+ if (deepclone) {
+ confirmation = (ChannelMessage) msg.deepclone();
} else {
- confirmation = (ChannelMessage)msg.clone();
+ confirmation = (ChannelMessage) msg.clone();
}
confirmation.getMessage().reset();
- UUIDGenerator.randomUUID(false,confirmation.getUniqueId(),0);
- confirmation.getMessage().append(START_DATA,0,START_DATA.length);
- confirmation.getMessage().append(msg.getUniqueId(),0,msg.getUniqueId().length);
- confirmation.getMessage().append(END_DATA,0,END_DATA.length);
- super.sendMessage(destination,confirmation,payload);
+ UUIDGenerator.randomUUID(false, confirmation.getUniqueId(), 0);
+ confirmation.getMessage().append(START_DATA, 0, START_DATA.length);
+ confirmation.getMessage().append(msg.getUniqueId(), 0, msg.getUniqueId().length);
+ confirmation.getMessage().append(END_DATA, 0, END_DATA.length);
+ super.sendMessage(destination, confirmation, payload);
} else {
- //turn off two phase commit
- //this won't work if the interceptor has 0 as a flag
- //since there is no flag to turn off
- //msg.setOptions(msg.getOptions() & (~getOptionFlag()));
+ // turn off two phase commit
+ // this won't work if the interceptor has 0 as a flag
+ // since there is no flag to turn off
+ // msg.setOptions(msg.getOptions() & (~getOptionFlag()));
super.sendMessage(destination, msg, payload);
}
}
@@ -74,21 +76,24 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase {
@Override
public void messageReceived(ChannelMessage msg) {
if (okToProcess(msg.getOptions())) {
- if ( msg.getMessage().getLength() == (START_DATA.length+msg.getUniqueId().length+END_DATA.length) &&
- Arrays.contains(msg.getMessage().getBytesDirect(),0,START_DATA,0,START_DATA.length) &&
- Arrays.contains(msg.getMessage().getBytesDirect(),START_DATA.length+msg.getUniqueId().length,END_DATA,0,END_DATA.length) ) {
- UniqueId id = new UniqueId(msg.getMessage().getBytesDirect(),START_DATA.length,msg.getUniqueId().length);
+ if (msg.getMessage().getLength() == (START_DATA.length + msg.getUniqueId().length + END_DATA.length) &&
+ Arrays.contains(msg.getMessage().getBytesDirect(), 0, START_DATA, 0, START_DATA.length) &&
+ Arrays.contains(msg.getMessage().getBytesDirect(), START_DATA.length + msg.getUniqueId().length,
+ END_DATA, 0, END_DATA.length)) {
+ UniqueId id =
+ new UniqueId(msg.getMessage().getBytesDirect(), START_DATA.length, msg.getUniqueId().length);
MapEntry original = messages.get(id);
- if ( original != null ) {
+ if (original != null) {
super.messageReceived(original.msg);
messages.remove(id);
} else {
- log.warn(sm.getString("twoPhaseCommitInterceptor.originalMessage.missing", Arrays.toString(id.getBytes())));
+ log.warn(sm.getString("twoPhaseCommitInterceptor.originalMessage.missing",
+ Arrays.toString(id.getBytes())));
}
} else {
UniqueId id = new UniqueId(msg.getUniqueId());
- MapEntry entry = new MapEntry((ChannelMessage)msg.deepclone(),id,System.currentTimeMillis());
- messages.put(id,entry);
+ MapEntry entry = new MapEntry((ChannelMessage) msg.deepclone(), id, System.currentTimeMillis());
+ messages.put(id, entry);
}
} else {
super.messageReceived(msg);
@@ -117,15 +122,15 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase {
long now = System.currentTimeMillis();
@SuppressWarnings("unchecked")
Map.Entry<UniqueId,MapEntry>[] entries = messages.entrySet().toArray(new Map.Entry[0]);
- for (Map.Entry<UniqueId, MapEntry> uniqueIdMapEntryEntry : entries) {
+ for (Map.Entry<UniqueId,MapEntry> uniqueIdMapEntryEntry : entries) {
MapEntry entry = uniqueIdMapEntryEntry.getValue();
if (entry.expired(now, expire)) {
log.info(sm.getString("twoPhaseCommitInterceptor.expiredMessage", entry.id));
messages.remove(entry.id);
}
}
- } catch ( Exception x ) {
- log.warn(sm.getString("twoPhaseCommitInterceptor.heartbeat.failed"),x);
+ } catch (Exception x) {
+ log.warn(sm.getString("twoPhaseCommitInterceptor.heartbeat.failed"), x);
} finally {
super.heartbeat();
}
@@ -141,8 +146,9 @@ public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase {
this.id = id;
this.timestamp = timestamp;
}
+
public boolean expired(long now, long expiration) {
- return (now - timestamp ) > expiration;
+ return (now - timestamp) > expiration;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org