You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2017/12/28 00:43:02 UTC
[geode] branch develop updated: GEODE-4163 clean up
DistributionManager code
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new 3be095a GEODE-4163 clean up DistributionManager code
3be095a is described below
commit 3be095af2baf90e7c2f2ee68c860031590bf944c
Author: Bruce Schuchardt <bs...@pivotal.io>
AuthorDate: Wed Dec 27 16:29:51 2017 -0800
GEODE-4163 clean up DistributionManager code
Removed all dead code - about 800 lines of code have been removed.
Removed DistributionChannel, which served no useful purpose.
Made many methods private or package-private.
Gathered most statics and instance variables scattered around in the file
and moved them to the top.
I attempted to replace use of InternalDistributedMember with
DistributedMember but the use of the former is too widespread. If we
want to do that it will change a large set of files.
This closes #1206
---
.../apache/geode/admin/jmx/internal/AgentImpl.java | 4 +-
.../geode/admin/jmx/internal/AgentLauncher.java | 2 +-
.../geode/distributed/DistributedSystem.java | 4 +-
.../org/apache/geode/distributed/internal/DM.java | 44 +-
.../distributed/internal/DistributionChannel.java | 162 ---
.../distributed/internal/DistributionManager.java | 1186 ++++++++------------
.../distributed/internal/DistributionMessage.java | 10 +-
.../internal/InternalDistributedSystem.java | 5 +-
.../internal/LonerDistributionManager.java | 13 +-
.../internal/StartupResponseMessage.java | 1 -
.../internal/membership/MembershipManager.java | 7 +
.../membership/gms/mgr/GMSMembershipManager.java | 9 +
.../admin/remote/RemoteGfManagerAgent.java | 2 +-
.../geode/internal/cache/GemFireCacheImpl.java | 2 +-
.../geode/internal/cache/PartitionedRegion.java | 2 +-
.../cache/SearchLoadAndWriteProcessor.java | 4 +-
.../org/apache/geode/admin/AdminTestHelper.java | 10 +-
.../geode/cache/ConnectionPoolDUnitTest.java | 6 +-
.../geode/distributed/SystemAdminDUnitTest.java | 3 +-
.../ConsoleDistributionManagerDUnitTest.java | 6 +-
.../InternalDistributedSystemJUnitTest.java | 3 +-
.../internal/membership/MembershipJUnitTest.java | 8 +-
.../gms/locator/GMSLocatorRecoveryJUnitTest.java | 2 +-
23 files changed, 524 insertions(+), 971 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/admin/jmx/internal/AgentImpl.java b/geode-core/src/main/java/org/apache/geode/admin/jmx/internal/AgentImpl.java
index 7d21ff7..c635f03 100644
--- a/geode-core/src/main/java/org/apache/geode/admin/jmx/internal/AgentImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/admin/jmx/internal/AgentImpl.java
@@ -411,7 +411,7 @@ public class AgentImpl implements org.apache.geode.admin.jmx.Agent,
return ((AdminDistributedSystemJmxImpl) this.system).getObjectName();
}
- DistributionManager.isDedicatedAdminVM = true;
+ DistributionManager.setIsDedicatedAdminVM(true);
AdminDistributedSystemJmxImpl systemJmx = (AdminDistributedSystemJmxImpl) this.system;
if (systemJmx == null) {
@@ -477,7 +477,7 @@ public class AgentImpl implements org.apache.geode.admin.jmx.Agent,
logger.warn(e.getMessage(), e);
throw e;
} finally {
- DistributionManager.isDedicatedAdminVM = false;
+ DistributionManager.setIsDedicatedAdminVM(false);
}
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/admin/jmx/internal/AgentLauncher.java b/geode-core/src/main/java/org/apache/geode/admin/jmx/internal/AgentLauncher.java
index e28c7a8..1819655 100644
--- a/geode-core/src/main/java/org/apache/geode/admin/jmx/internal/AgentLauncher.java
+++ b/geode-core/src/main/java/org/apache/geode/admin/jmx/internal/AgentLauncher.java
@@ -376,7 +376,7 @@ public class AgentLauncher {
}
private Agent createAgent(final Properties props) throws IOException, AdminException {
- DistributionManager.isDedicatedAdminVM = true;
+ DistributionManager.setIsDedicatedAdminVM(true);
SystemFailure.setExitOK(true);
final AgentConfigImpl config = new AgentConfigImpl(props);
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/DistributedSystem.java b/geode-core/src/main/java/org/apache/geode/distributed/DistributedSystem.java
index d740bd3..8c24214 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/DistributedSystem.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/DistributedSystem.java
@@ -156,7 +156,7 @@ public abstract class DistributedSystem implements StatisticsFactory {
config = new Properties();
}
synchronized (existingSystemsLock) {
- if (DistributionManager.isDedicatedAdminVM) {
+ if (DistributionManager.isDedicatedAdminVM()) {
// For a dedicated admin VM, check to see if there is already
// a connect that will suit our purposes.
DistributedSystem existingSystem = getConnection(config);
@@ -323,7 +323,7 @@ public abstract class DistributedSystem implements StatisticsFactory {
LocalizedStrings.DistributedSystem_THIS_VM_ALREADY_HAS_ONE_OR_MORE_DISTRIBUTED_SYSTEM_CONNECTIONS_0
.toLocalizedString(existingSystems));
}
- DistributionManager.isDedicatedAdminVM = adminOnly;
+ DistributionManager.setIsDedicatedAdminVM(adminOnly);
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/DM.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/DM.java
index 31f2154..f55d61a 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/DM.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/DM.java
@@ -21,6 +21,7 @@ import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import org.apache.geode.CancelCriterion;
@@ -175,27 +176,15 @@ public interface DM extends ReplySender {
ElderState getElderState(boolean force, boolean useTryLock);
/**
- * Returns the id of the underlying distribution channel used for communication.
+ * Returns the membership port of the underlying distribution manager used for communication.
*
* @since GemFire 3.0
*/
- long getChannelId();
+ long getMembershipPort();
/**
- * Adds a message to the outgoing queue. Note that <code>message</code> should not be modified
- * after it has been added to the queue. After <code>message</code> is distributed, it will be
- * recycled.
+ * Sends a message
*
- * @return recipients who did not receive the message
- * @throws NotSerializableException If <code>message</code> cannot be serialized
- * @see #putOutgoing(DistributionMessage)
- */
- Set putOutgoingUserData(DistributionMessage message) throws NotSerializableException;
-
- /**
- * Sends a message, guaranteed to be serialized
- *
- * @see #putOutgoingUserData(DistributionMessage)
* @param msg
* @return recipients who did not receive the message
*/
@@ -227,6 +216,11 @@ public interface DM extends ReplySender {
*/
void removeAllMembershipListener(MembershipListener l);
+ /**
+ * Makes note of a new administration console (admin-only member).
+ *
+ * @Deprecated admin members are deprecated
+ */
void addAdminConsole(InternalDistributedMember id);
DMStats getStats();
@@ -269,6 +263,11 @@ public interface DM extends ReplySender {
ExecutorService getPrMetaDataCleanupThreadPool();
/**
+ * Return the executor used for function processing
+ */
+ Executor getFunctionExcecutor();
+
+ /**
* gets this distribution manager's message-processing executor for ordered (i.e. serialized)
* message processing
*/
@@ -295,7 +294,7 @@ public interface DM extends ReplySender {
/**
* @return Set of Admin VM nodes
*/
- Set getAdminMemberSet();
+ Set<InternalDistributedMember> getAdminMemberSet();
/** Throws ShutdownException if closeInProgress returns true. */
void throwIfDistributionStopped();
@@ -440,14 +439,6 @@ public interface DM extends ReplySender {
*/
Map<InternalDistributedMember, Collection<String>> getAllHostedLocatorsWithSharedConfiguration();
- /****
- * Determines if the distributed system has the shared configuration service enabled or not.
- *
- * @return true if the distributed system was started or had a locator with
- * enable-cluster-configuration = true
- */
- boolean isSharedConfigurationServiceEnabledForDS();
-
/**
* Forces use of UDP for communications in the current thread. UDP is connectionless, so no tcp/ip
* connections will be created or used for messaging until this setting is released with
@@ -461,6 +452,11 @@ public interface DM extends ReplySender {
*/
void releaseUDPMessagingForCurrentThread();
+ /**
+ * returns the type of node
+ *
+ * @return
+ */
int getDMType();
/**
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionChannel.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionChannel.java
deleted file mode 100644
index 89501db..0000000
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionChannel.java
+++ /dev/null
@@ -1,162 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
- * agreements. See the NOTICE file distributed with this work for additional information regarding
- * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance with the License. You may obtain a
- * copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- */
-package org.apache.geode.distributed.internal;
-
-import java.io.NotSerializableException;
-import java.util.HashSet;
-import java.util.Set;
-
-import org.apache.logging.log4j.Logger;
-
-import org.apache.geode.distributed.DistributedSystemDisconnectedException;
-import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
-import org.apache.geode.distributed.internal.membership.MembershipManager;
-import org.apache.geode.internal.i18n.LocalizedStrings;
-import org.apache.geode.internal.logging.LogService;
-import org.apache.geode.internal.logging.log4j.LocalizedMessage;
-import org.apache.geode.internal.logging.log4j.LogMarker;
-
-/**
- * To change this generated comment edit the template variable "typecomment":
- * Window>Preferences>Java>Templates. To enable and disable the creation of type comments go to
- * Window>Preferences>Java>Code Generation.
- */
-public class DistributionChannel {
- private static final Logger logger = LogService.getLogger();
-
- private MembershipManager membershipManager;
-
- /**
- * Constructor DistributionChannel for JGroups.
- *
- * @param channel jgroups channel
- */
- public DistributionChannel(MembershipManager channel) {
- membershipManager = channel;
- }
-
-
- public InternalDistributedMember getLocalAddress() {
- return membershipManager.getLocalMember();
- }
-
-
- /**
- * @return the MembershipManager
- */
- public MembershipManager getMembershipManager() {
- return membershipManager;
- }
-
-
-
- /**
- * @return list of recipients who did not receive the message because they left the view (null if
- * all received it or it was sent to {@link DistributionMessage#ALL_RECIPIENTS}).
- * @throws NotSerializableException If content cannot be serialized
- */
- public Set send(InternalDistributedMember[] destinations, DistributionMessage content,
- DistributionManager dm, DistributionStats stats) throws NotSerializableException {
- if (membershipManager == null) {
- logger.warn(LocalizedMessage.create(
- LocalizedStrings.DistributionChannel_ATTEMPTING_A_SEND_TO_A_DISCONNECTED_DISTRIBUTIONMANAGER));
- if (destinations.length == 1 && destinations[0] == DistributionMessage.ALL_RECIPIENTS)
- return null;
- HashSet result = new HashSet();
- for (int i = 0; i < destinations.length; i++)
- result.add(destinations[i]);
- return result;
- }
- return membershipManager.send(destinations, content, stats);
- }
-
- public void disconnect(boolean beforeJoined) {
- StringBuffer sb = new StringBuffer();
- sb.append("Disconnected from distribution channel ");
-
- long start = System.currentTimeMillis();
-
- logger.debug("DistributionChannel disconnecting with " + membershipManager + "; beforeJoined="
- + beforeJoined);
-
- if (membershipManager != null) {
- sb.append(membershipManager.getLocalMember());
- sb.append(" (took ");
- long begin = System.currentTimeMillis();
- if (beforeJoined) {
- membershipManager.uncleanShutdown("Failed to start distribution", null);
- } else {
- membershipManager.shutdown();
- }
- long delta = System.currentTimeMillis() - begin;
- sb.append(delta);
- sb.append("/");
- }
- membershipManager = null;
-
- if (logger.isTraceEnabled(LogMarker.DM)) {
- long delta = System.currentTimeMillis() - start;
- sb.append(delta);
- sb.append(" ms)");
- logger.trace(LogMarker.DM, sb);
- }
- }
-
- /**
- * Returns the id of this distribution channel. If this channel uses JavaGroups and the conduit to
- * communicate with others, then the port of the JavaGroups channel's
- * {@link InternalDistributedMember address} is returned.
- *
- * @since GemFire 3.0
- */
- public long getId() {
- MembershipManager mgr = this.membershipManager;
- if (mgr == null) {
- throw new DistributedSystemDisconnectedException(
- LocalizedStrings.DistributionChannel_I_NO_LONGER_HAVE_A_MEMBERSHIP_ID
- .toLocalizedString());
- }
- InternalDistributedMember moi = mgr.getLocalMember();
- if (moi == null) {
- throw new DistributedSystemDisconnectedException(
- LocalizedStrings.DistributionChannel_I_NO_LONGER_HAVE_A_MEMBERSHIP_ID.toLocalizedString(),
- membershipManager.getShutdownCause());
- }
- return moi.getPort();
- }
-
- public void setShutDown() {
- // this.shuttingDown = shuttingDown;
- if (membershipManager != null)
- membershipManager.setShutdown();
- }
-
- // private void sendViaJGroups(Serializable[] destinations,Address source,Serializable content,
- // boolean deliverToSender, int processorType,
- // DistributionManager dm)
- // throws ChannelNotConnectedException, ChannelClosedException {
- // Message msg = new Message(null, source, content);
- // msg.setDeliverToSender(deliverToSender);
- // msg.setProcessorType(processorType);
- // for (int i=0; i < destinations.length; i++) {
- // Address destination = (Address) destinations[i];
- // msg.setDest(destination);
- // jgroupsChannel.send(msg);
- // if (destination == null)
- // break;
- // }
- // }
-
-}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionManager.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionManager.java
index 6130e12..5a4de3d 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionManager.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionManager.java
@@ -90,7 +90,10 @@ import org.apache.geode.internal.util.concurrent.StoppableReentrantLock;
/**
* The <code>DistributionManager</code> uses a {@link MembershipManager} to distribute
- * {@link DistributionMessage messages} queued in {@link MQueue}s.
+ * {@link DistributionMessage messages}. It also reports on who is currently in the distributed
+ * system and tracks the elder member for the distributed lock service. You may also register a
+ * membership listener with the DistributionManager to receive notification of changes in
+ * membership.
*
* <P>
*
@@ -99,15 +102,6 @@ import org.apache.geode.internal.util.concurrent.StoppableReentrantLock;
*
* <P>
*
- * Prior to GemFire 4.0, <code>DistributionManager</code> was an abstract class with two concrete
- * subclasses, <code>LocalDistributionManager</code> and <code>ConsoleDistributionManager</code>. We
- * decided that <code>ConsoleDistributionManager</code> (which was used for the GUI console and
- * admin APIs) did not offer enough interesting functionality to warrant a separate class. More
- * importantly, it prevented the Cache and admin APIs from being used in the same VM. So, we
- * refactored the code of those two subclasses into <code>DistributionManager</code>.
- *
- * @since GemFire 2.0
- *
* @see DistributionMessage#process
* @see IgnoredByManager
*/
@@ -115,150 +109,129 @@ public class DistributionManager implements DM {
private static final Logger logger = LogService.getLogger();
- private static final boolean SYNC_EVENTS = Boolean.getBoolean("DistributionManager.syncEvents");
+ private static final int STARTUP_TIMEOUT =
+ Integer.getInteger("DistributionManager.STARTUP_TIMEOUT", 15000).intValue();
+
+ public static final boolean DEBUG_NO_ACKNOWLEDGEMENTS =
+ Boolean.getBoolean("DistributionManager.DEBUG_NO_ACKNOWLEDGEMENTS");
/**
- * WARNING: setting this to true may break dunit tests.
- * <p>
- * see org.apache.geode.cache30.ClearMultiVmCallBkDUnitTest
+ * maximum time, in milliseconds, to wait for all threads to exit
*/
- public static final boolean INLINE_PROCESS =
- !Boolean.getBoolean("DistributionManager.enqueueOrderedMessages");
+ private static final int MAX_STOP_TIME = 20000;
/**
- * Flag indicating whether to use single Serial-Executor thread or Multiple Serial-executor
- * thread,
+ * Time to sleep, in milliseconds, while polling to see if threads have finished
*/
- public static final boolean MULTI_SERIAL_EXECUTORS =
- !Boolean.getBoolean("DistributionManager.singleSerialExecutor");
-
- /** The name of the distribution manager (identifies it in GemFire) */
- public static final String NAME = "GemFire";
+ private static final int STOP_PAUSE_TIME = 1000;
/**
- * The number of milliseconds to wait for distribution-related things to happen
+ * Maximum number of interrupt attempts to stop a thread
*/
- public static final long TIMEOUT = Long.getLong("DistributionManager.TIMEOUT", -1).longValue();
+ private static final int MAX_STOP_ATTEMPTS = 10;
- public static final int PUSHER_THREADS =
- Integer.getInteger("DistributionManager.PUSHER_THREADS", 50).intValue();
- public static final int PUSHER_QUEUE_SIZE =
- Integer.getInteger("DistributionManager.PUSHER_QUEUE_SIZE", 4096).intValue();
+ private static final boolean SYNC_EVENTS = Boolean.getBoolean("DistributionManager.syncEvents");
+
+ /**
+ * Flag indicating whether to use single Serial-Executor thread or Multiple Serial-executor
+ * thread,
+ */
+ private static final boolean MULTI_SERIAL_EXECUTORS =
+ !Boolean.getBoolean("DistributionManager.singleSerialExecutor");
- public static final int MAX_WAITING_THREADS =
+ private static final int MAX_WAITING_THREADS =
Integer.getInteger("DistributionManager.MAX_WAITING_THREADS", Integer.MAX_VALUE).intValue();
- public static final int MAX_PR_META_DATA_CLEANUP_THREADS =
+ private static final int MAX_PR_META_DATA_CLEANUP_THREADS =
Integer.getInteger("DistributionManager.MAX_PR_META_DATA_CLEANUP_THREADS", 1).intValue();
public static final int MAX_THREADS =
Integer.getInteger("DistributionManager.MAX_THREADS", 100).intValue();
- public static final int MAX_PR_THREADS = Integer.getInteger("DistributionManager.MAX_PR_THREADS",
+
+ private static final int MAX_PR_THREADS = Integer.getInteger("DistributionManager.MAX_PR_THREADS",
Math.max(Runtime.getRuntime().availableProcessors() * 4, 16)).intValue();
- public static final int MAX_FE_THREADS = Integer.getInteger("DistributionManager.MAX_FE_THREADS",
+
+ private static final int MAX_FE_THREADS = Integer.getInteger("DistributionManager.MAX_FE_THREADS",
Math.max(Runtime.getRuntime().availableProcessors() * 4, 16)).intValue();
- // Integer.getInteger("DistributionManager.MAX_THREADS",
- // max(Runtime.getRuntime().availableProcessors()*2, 2)).intValue();
- public static final int INCOMING_QUEUE_LIMIT =
+
+
+ private static final int INCOMING_QUEUE_LIMIT =
Integer.getInteger("DistributionManager.INCOMING_QUEUE_LIMIT", 80000).intValue();
- public static final int INCOMING_QUEUE_THROTTLE =
- Integer.getInteger("DistributionManager.INCOMING_QUEUE_THROTTLE",
- (int) (INCOMING_QUEUE_LIMIT * 0.75)).intValue();
/** Throttling based on the Queue byte size */
- public static final double THROTTLE_PERCENT = (double) (Integer
+ private static final double THROTTLE_PERCENT = (double) (Integer
.getInteger("DistributionManager.SERIAL_QUEUE_THROTTLE_PERCENT", 75).intValue()) / 100;
- public static final int SERIAL_QUEUE_BYTE_LIMIT = Integer
+
+ static final int SERIAL_QUEUE_BYTE_LIMIT = Integer
.getInteger("DistributionManager.SERIAL_QUEUE_BYTE_LIMIT", (40 * (1024 * 1024))).intValue();
- public static final int SERIAL_QUEUE_THROTTLE =
+
+ static final int SERIAL_QUEUE_THROTTLE =
Integer.getInteger("DistributionManager.SERIAL_QUEUE_THROTTLE",
(int) (SERIAL_QUEUE_BYTE_LIMIT * THROTTLE_PERCENT)).intValue();
- public static final int TOTAL_SERIAL_QUEUE_BYTE_LIMIT =
+
+ static final int TOTAL_SERIAL_QUEUE_BYTE_LIMIT =
Integer.getInteger("DistributionManager.TOTAL_SERIAL_QUEUE_BYTE_LIMIT", (80 * (1024 * 1024)))
.intValue();
- public static final int TOTAL_SERIAL_QUEUE_THROTTLE =
+
+ static final int TOTAL_SERIAL_QUEUE_THROTTLE =
Integer.getInteger("DistributionManager.TOTAL_SERIAL_QUEUE_THROTTLE",
(int) (SERIAL_QUEUE_BYTE_LIMIT * THROTTLE_PERCENT)).intValue();
/** Throttling based on the Queue item size */
- public static final int SERIAL_QUEUE_SIZE_LIMIT =
+ static final int SERIAL_QUEUE_SIZE_LIMIT =
Integer.getInteger("DistributionManager.SERIAL_QUEUE_SIZE_LIMIT", 20000).intValue();
- public static final int SERIAL_QUEUE_SIZE_THROTTLE =
+
+ static final int SERIAL_QUEUE_SIZE_THROTTLE =
Integer.getInteger("DistributionManager.SERIAL_QUEUE_SIZE_THROTTLE",
(int) (SERIAL_QUEUE_SIZE_LIMIT * THROTTLE_PERCENT)).intValue();
/** Max number of serial Queue executors, in case of multi-serial-queue executor */
- public static final int MAX_SERIAL_QUEUE_THREAD =
+ static final int MAX_SERIAL_QUEUE_THREAD =
Integer.getInteger("DistributionManager.MAX_SERIAL_QUEUE_THREAD", 20).intValue();
- /**
- * Whether or not to include link local addresses in the list of addresses we use to determine if
- * two members are no the same host.
- *
- * Added for normura issue 7033 - they have duplicate link local addresses on different boxes
- */
- public static volatile boolean INCLUDE_LINK_LOCAL_ADDRESSES =
- Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "IncludeLinkLocalAddresses");
+
/** The DM type for regular distribution managers */
public static final int NORMAL_DM_TYPE = 10;
- /**
- * The DM type for locator distribution managers
- *
- * @since GemFire 7.0
- */
+ /** The DM type for locator distribution managers */
public static final int LOCATOR_DM_TYPE = 11;
/** The DM type for Console (admin-only) distribution managers */
public static final int ADMIN_ONLY_DM_TYPE = 12;
+ /** The DM type for stand-alone members */
public static final int LONER_DM_TYPE = 13;
+
+
/**
- * an NIO priority type
- *
* @see org.apache.geode.distributed.internal.PooledDistributionMessage
- * @see #SERIAL_EXECUTOR
- * @see #HIGH_PRIORITY_EXECUTOR
- * @see #WAITING_POOL_EXECUTOR
*/
public static final int STANDARD_EXECUTOR = 73;
/**
- * an NIO priority type
- *
* @see org.apache.geode.distributed.internal.SerialDistributionMessage
- * @see #STANDARD_EXECUTOR
*/
public static final int SERIAL_EXECUTOR = 74;
/**
- * an NIO priority type
- *
* @see org.apache.geode.distributed.internal.HighPriorityDistributionMessage
- * @see #STANDARD_EXECUTOR
*/
public static final int HIGH_PRIORITY_EXECUTOR = 75;
// 76 not in use
/**
- * an NIO priority type
- *
* @see org.apache.geode.internal.cache.InitialImageOperation
- * @see #STANDARD_EXECUTOR
*/
public static final int WAITING_POOL_EXECUTOR = 77;
/**
- * an NIO priority type
- *
* @see org.apache.geode.internal.cache.InitialImageOperation
- * @see #STANDARD_EXECUTOR
*/
public static final int PARTITIONED_REGION_EXECUTOR = 78;
@@ -267,70 +240,91 @@ public class DistributionManager implements DM {
* Executor for view related messages
*
* @see org.apache.geode.distributed.internal.membership.gms.messages.ViewAckMessage
- * @see #STANDARD_EXECUTOR
*/
public static final int VIEW_EXECUTOR = 79;
public static final int REGION_FUNCTION_EXECUTION_EXECUTOR = 80;
- /** The number of open distribution managers in this VM */
- private static int openDMs = 0;
- // /** The stack trace of the last time a console DM was opened */
- // private static Exception openStackTrace;
+
+ /** Is this node running an AdminDistributedSystem? */
+ private static volatile boolean isDedicatedAdminVM = false;
+
+ private static ThreadLocal<Boolean> isStartupThread = new ThreadLocal();
/**
- * Is this VM dedicated to administration (like a GUI console or a JMX agent)? If so, then it
- * creates {@link #ADMIN_ONLY_DM_TYPE} type distribution managers.
- *
- * @since GemFire 4.0
+ * Identifier for function execution threads and any of their children
*/
- public static volatile boolean isDedicatedAdminVM = false;
+ private static final InheritableThreadLocal<Boolean> isFunctionExecutionThread =
+ new InheritableThreadLocal<Boolean>() {
+ @Override
+ protected Boolean initialValue() {
+ return Boolean.FALSE;
+ }
+ };
+
+
+
+ ///////////////////// Instance Fields //////////////////////
/**
- * Is this admin agent used for a command line console. This flag controls whether connect will
- * throw an exception or just wait for a DS if one is not available. If true, we will throw an
- * exception.
- *
+ * Mutex to control access to {@link #waitingForElderChange} or {@link #elder}.
*/
- public static volatile boolean isCommandLineAdminVM = false;
+ private final Object elderMonitor = new Object();
+ /**
+ * Must be read/written while holding {@link #elderMonitor}
+ *
+ * @see #elderChangeWait()
+ */
+ private boolean waitingForElderChange = false;
+ /**
+ * @see DM#isAdam()
+ */
+ private boolean adam = false;
- ///////////////////// Instance Fields //////////////////////
+ /**
+ * This is the "elder" member of the distributed system, responsible for certain types of
+ * arbitration.
+ *
+ * Must hold {@link #elderMonitor} in order to change this.
+ *
+ * @see #getElderId()
+ */
+ protected volatile InternalDistributedMember elder = null;
/** The id of this distribution manager */
- protected final InternalDistributedMember myid;
+ protected final InternalDistributedMember localAddress;
/**
* The distribution manager type of this dm; set in its constructor.
- * </p>
- * TODO: change this to use an enum
*/
private final int dmType;
/**
* The <code>MembershipListener</code>s that are registered on this manager.
*/
- private final ConcurrentMap membershipListeners;
+ private final ConcurrentMap<MembershipListener, Boolean> membershipListeners;
- /** A lock to hold while adding and removing membership listeners */
- protected final Object membershipListenersLock = new MembershipListenersLock();
/**
* The <code>MembershipListener</code>s that are registered on this manager for ALL members.
*
* @since GemFire 5.7
*/
- protected volatile Set allMembershipListeners = Collections.EMPTY_SET;
+ protected volatile Set<MembershipListener> allMembershipListeners = Collections.emptySet();
+
/**
* A lock to hold while adding and removing all membership listeners.
*
* @since GemFire 5.7
*/
- protected final Object allMembershipListenersLock = new MembershipListenersLock();
+ private final Object allMembershipListenersLock = new MembershipListenersLock();
+
/** A queue of MemberEvent instances */
- protected final BlockingQueue membershipEventQueue = new LinkedBlockingQueue();
+ private final BlockingQueue<MemberEvent> membershipEventQueue = new LinkedBlockingQueue();
+
/** Used to invoke registered membership listeners in the background. */
private Thread memberEventThread;
@@ -339,26 +333,21 @@ public class DistributionManager implements DM {
protected final String description;
/** Statistics about distribution */
- protected /* final */ DistributionStats stats;
+ protected DistributionStats stats;
/** Did an exception occur in one of the DM threads? */
- protected boolean exceptionInThreads;
-
- static ThreadLocal isStartupThread = new ThreadLocal();
+ private boolean exceptionInThreads;
- protected volatile boolean shutdownMsgSent = false;
+ private volatile boolean shutdownMsgSent = false;
/** Set to true when this manager is being shutdown */
- protected volatile boolean closeInProgress = false;
+ private volatile boolean closeInProgress = false;
private volatile boolean receivedStartupResponse = false;
private volatile String rejectionMessage = null;
- protected MembershipManager membershipManager;
-
- /** The channel through which distributed communication occurs. */
- protected DistributionChannel channel;
+ private MembershipManager membershipManager;
/**
* The (non-admin-only) members of the distributed system. This is a map of memberid->memberid for
@@ -371,7 +360,7 @@ public class DistributionManager implements DM {
* All (admin and non-admin) members of the distributed system. All accesses to this field must be
* synchronized on {@link #membersLock}.
*/
- private Set membersAndAdmin = Collections.emptySet();
+ private Set<InternalDistributedMember> membersAndAdmin = Collections.emptySet();
/**
* Map of all locator members of the distributed system. The value is a collection of locator
* strings that are hosted in that member. All accesses to this field must be synchronized on
@@ -389,11 +378,6 @@ public class DistributionManager implements DM {
Collections.emptyMap();
/**
- * Since 6.6.2 and hereafter we will save the versions here. But pre-6.6.2's
- * StartupResponseMessage does not contain version. We will assign a default version for them.
- */
- public static final String DEFAULT_VERSION_PRE_6_6_2 = "6.6.0.0";
- /**
* The lock held while accessing the field references to the following:<br>
* 1) {@link #members}<br>
* 2) {@link #membersAndAdmin}<br>
@@ -412,9 +396,6 @@ public class DistributionManager implements DM {
*/
private volatile Set<InternalDistributedMember> adminConsoles = Collections.emptySet();
- /** The pusher thread */
- // private Thread pusher;
-
/** The group of distribution manager threads */
protected LoggingThreadGroup threadGroup;
@@ -445,6 +426,8 @@ public class DistributionManager implements DM {
*/
private ThreadPoolExecutor partitionedRegionThread;
private ThreadPoolExecutor partitionedRegionPool;
+
+ /** Function Execution executors */
private ThreadPoolExecutor functionExecutionThread;
private ThreadPoolExecutor functionExecutionPool;
@@ -467,7 +450,7 @@ public class DistributionManager implements DM {
/** a map keyed on InternalDistributedMember, to direct channels to other systems */
// protected final Map channelMap = CFactory.createCM();
- protected volatile boolean readyForMessages = false;
+ private volatile boolean readyForMessages = false;
/**
* Set to true once this DM is ready to send messages. Note that it is always ready to send the
@@ -477,7 +460,7 @@ public class DistributionManager implements DM {
private final Object readyToSendMsgsLock = new Object();
/** Is this distribution manager closed? */
- protected volatile boolean closed = false;
+ private volatile boolean closed = false;
/**
* The distributed system to which this distribution manager is connected.
@@ -494,14 +477,17 @@ public class DistributionManager implements DM {
private SerialQueuedExecutorPool serialQueuedExecutorPool;
+ /**
+ * TODO why does the distribution manager arbitrate GII operations? That should be a Cache
+ * function
+ */
private final Semaphore parallelGIIs = new Semaphore(InitialImageOperation.MAX_PARALLEL_GIIS);
/**
* Map of InetAddress to HashSets of InetAddress, to define equivalences between network interface
* cards and hosts.
*/
- private final HashMap<InetAddress, Set<InetAddress>> equivalentHosts =
- new HashMap<InetAddress, Set<InetAddress>>();
+ private final HashMap<InetAddress, Set<InetAddress>> equivalentHosts = new HashMap<>();
private int distributedSystemId = DistributionConfig.DEFAULT_DISTRIBUTED_SYSTEM_ID;
@@ -511,34 +497,25 @@ public class DistributionManager implements DM {
private boolean enforceUniqueZone = false;
- private volatile boolean isSharedConfigEnabledForDS = false;
-
- @Override
- public boolean isSharedConfigurationServiceEnabledForDS() {
- return isSharedConfigEnabledForDS;
- }
+ /**
+ * root cause of forcibly shutting down the distribution manager
+ */
+ volatile Throwable rootCause = null;
/**
- * Identifier for function execution threads and any of their children
+ * @see #closeInProgress
*/
- public static final InheritableThreadLocal<Boolean> isFunctionExecutionThread =
- new InheritableThreadLocal<Boolean>() {
- @Override
- protected Boolean initialValue() {
- return Boolean.FALSE;
- }
- };
+ private final Object shutdownMutex = new Object();
+
+
+
////////////////////// Static Methods //////////////////////
/**
- * Given two DistributionManager ids, check to see if they are from the same host address.
- *
- * @param id1 a DistributionManager id
- * @param id2 a DistributionManager id
- * @return true if id1 and id2 are from the same host, false otherwise
+ * Is the current thread used for executing Functions?
*/
- public static boolean isSameHost(InternalDistributedMember id1, InternalDistributedMember id2) {
- return (id1.getInetAddress().equals(id2.getInetAddress()));
+ public static Boolean isFunctionExecutionThread() {
+ return isFunctionExecutionThread.get();
}
/**
@@ -547,7 +524,7 @@ public class DistributionManager implements DM {
*
* @param system The distributed system to which this distribution manager will send messages.
*/
- public static DistributionManager create(InternalDistributedSystem system) {
+ static DistributionManager create(InternalDistributedSystem system) {
DistributionManager distributionManager = null;
boolean beforeJoined = true;
@@ -560,7 +537,7 @@ public class DistributionManager implements DM {
// if this DM is starting for a locator, set it to be a locator DM
vmKind = LOCATOR_DM_TYPE;
- } else if (isDedicatedAdminVM) {
+ } else if (isDedicatedAdminVM()) {
vmKind = ADMIN_ONLY_DM_TYPE;
} else {
@@ -608,7 +585,7 @@ public class DistributionManager implements DM {
StartupOperation op = new StartupOperation(distributionManager, transport);
try {
- if (!distributionManager.sendStartupMessage(op, true)) {
+ if (!distributionManager.sendStartupMessage(op)) {
// We'll we didn't hear back from anyone else. We assume that
// we're the first one.
if (distributionManager.getOtherDistributionManagerIds().size() == 0) {
@@ -663,127 +640,6 @@ public class DistributionManager implements DM {
}
}
- void runUntilShutdown(Runnable r) {
- try {
- r.run();
- } catch (CancelException e) {
- if (logger.isTraceEnabled()) {
- logger.trace("Caught shutdown exception", e);
- }
- } catch (VirtualMachineError err) {
- SystemFailure.initiateFailure(err);
- // If this ever returns, rethrow the error. We're poisoned
- // now, so don't let this thread continue.
- throw err;
- } catch (Throwable t) {
- // Whenever you catch Error or Throwable, you must also
- // catch VirtualMachineError (see above). However, there is
- // _still_ a possibility that you are dealing with a cascading
- // error condition, so you also need to check to see if the JVM
- // is still usable:
- SystemFailure.checkFailure();
- if (isCloseInProgress()) {
- logger.debug("Caught unusual exception during shutdown: {}", t.getMessage(), t);
- } else {
- logger.warn(LocalizedMessage
- .create(LocalizedStrings.DistributionManager_TASK_FAILED_WITH_EXCEPTION), t);
- }
- }
- }
-
- volatile Throwable rootCause = null;
-
- private static class Stopper extends CancelCriterion {
- private DistributionManager dm;
-
- // validateDM is commented out because expiry threads hit it with
- // an ugly failure... use only for debugging lingering DM bugs
- // private String validateDM() {
- // GemFireCache cache = GemFireCache.getInstance();
- // if (cache == null) {
- // return null; // Distributed system with no cache
- // }
- // Object obj = cache.getDistributedSystem();
- // if (obj == null) {
- // return null; // Cache is very dead
- // }
- // InternalDistributedSystem ids = (InternalDistributedSystem)obj;
- // DM current = ids.getDistributionManager();
- // if (current != dm) {
- // String response =
- // LocalizedStrings.DistributionManager_CURRENT_CACHE_DISTRIBUTIONMANAGER_0_IS_NOT_THE_SAME_AS_1
- // .toLocalizedString(new Object[] { current, dm});
- // return response;
- // }
- // return null;
- // }
-
- Stopper(DistributionManager dm) {
- this.dm = dm;
- }
-
- @Override
- public String cancelInProgress() {
- checkFailure();
-
- // remove call to validateDM() to fix bug 38356
-
- if (dm.shutdownMsgSent) {
- return LocalizedStrings.DistributionManager__0_MESSAGE_DISTRIBUTION_HAS_TERMINATED
- .toLocalizedString(dm.toString());
- }
- if (dm.rootCause != null) {
- return dm.toString() + ": " + dm.rootCause.getMessage();
- }
-
- // Nope.
- return null;
- }
-
- @Override
- public RuntimeException generateCancelledException(Throwable e) {
- String reason = cancelInProgress();
- if (reason == null) {
- return null;
- }
- Throwable rc = dm.rootCause; // volatile read
- if (rc == null) {
- // No root cause, specify the one given and be done with it.
- return new DistributedSystemDisconnectedException(reason, e);
- }
-
- if (e == null) {
- // Caller did not specify any root cause, so just use our own.
- return new DistributedSystemDisconnectedException(reason, rc);
- }
-
- // Attempt to stick rootCause at tail end of the exception chain.
- Throwable nt = e;
- while (nt.getCause() != null) {
- nt = nt.getCause();
- }
- if (nt == rc) {
- // Root cause already in place; we're done
- return new DistributedSystemDisconnectedException(reason, e);
- }
-
- try {
- nt.initCause(rc);
- return new DistributedSystemDisconnectedException(reason, e);
- } catch (IllegalStateException e2) {
- // Bug 39496 (Jrockit related) Give up. The following
- // error is not entirely sane but gives the correct general picture.
- return new DistributedSystemDisconnectedException(reason, rc);
- }
- }
- }
-
- private final Stopper stopper = new Stopper(this);
-
- public CancelCriterion getCancelCriterion() {
- return stopper;
- }
-
/////////////////////// Constructors ///////////////////////
/**
@@ -806,7 +662,7 @@ public class DistributionManager implements DM {
long statId = OSProcess.getId();
/*
* deadcoded since we don't know the channel id yet. if (statId == 0 || statId == -1) { statId
- * = getChannelId(); }
+ * = getMembershipPort(); }
*/
this.stats = new DistributionStats(system, statId);
DistributionStats.enableClockStats = system.getConfig().getEnableTimeStatistics();
@@ -1144,50 +1000,26 @@ public class DistributionManager implements DM {
StringBuffer sb = new StringBuffer(" (took ");
+ // connect to the cluster
long start = System.currentTimeMillis();
- // Create direct channel first
- // DirectChannel dc = new DirectChannel(new MyListener(this), system.getConfig(), logger,
- // null);
- // setDirectChannelPort(dc.getPort()); // store in a thread local
-
- // connect to JGroups
- start = System.currentTimeMillis();
-
- MyListener l = new MyListener(this);
+ DMListener l = new DMListener(this);
membershipManager = MemberFactory.newMembershipManager(l, system.getConfig(), transport,
stats, system.getSecurityService());
sb.append(System.currentTimeMillis() - start);
- this.myid = membershipManager.getLocalMember();
-
- // dc.patchUpAddress(this.myid);
- // id.setDirectChannelPort(dc.getPort());
-
- // create the distribution channel
- this.channel = new DistributionChannel(membershipManager);
+ this.localAddress = membershipManager.getLocalMember();
membershipManager.postConnect();
- // Assert.assertTrue(this.getChannelMap().size() >= 1);
- // System.out.println("Channel Map:");
- // for (Iterator iter = this.getChannelMap().entrySet().iterator();
- // iter.hasNext(); ) {
- // Map.Entry entry = (Map.Entry) iter.next();
- // Object key = entry.getKey();
- // System.out.println(" " + key + " a " +
- // key.getClass().getName() + " -> " +
- // entry.getValue());
- // }
-
sb.append(" ms)");
logger.info(LocalizedMessage.create(
- LocalizedStrings.DistributionManager_STARTING_DISTRIBUTIONMANAGER_0_1,
- new Object[] {this.myid, (logger.isInfoEnabled(LogMarker.DM) ? sb.toString() : "")}));
+ LocalizedStrings.DistributionManager_STARTING_DISTRIBUTIONMANAGER_0_1, new Object[] {
+ this.localAddress, (logger.isInfoEnabled(LogMarker.DM) ? sb.toString() : "")}));
- this.description = NAME + " on " + this.myid + " started at "
+ this.description = "Distribution manager on " + this.localAddress + " started at "
+ (new Date(System.currentTimeMillis())).toString();
finishedConstructor = true;
@@ -1209,25 +1041,10 @@ public class DistributionManager implements DM {
boolean finishedConstructor = false;
try {
- isStartupThread.set(Boolean.TRUE);
+ setIsStartupThread(Boolean.TRUE);
startThreads();
- // Since we need a StartupResponseMessage to make sure licenses
- // are compatible the following has been deadcoded.
- // // For the time being, invoke processStartupResponse()
- // String rejectionMessage = null;
- // if (GemFireVersion.getGemFireVersion().
- // equals(state.getGemFireVersion())) {
- // rejectionMessage = "Rejected new system node " +
- // this.getDistributionManagerId() + " with version \"" +
- // GemFireVersion.getGemFireVersion() +
- // "\" because the distributed system's version is \"" +
- // state.getGemFireVersion() + "\".";
- // }
- // this.processStartupResponse(state.getCacheTime(),
- // rejectionMessage);
-
// Allow events to start being processed.
membershipManager.startEventProcessing();
for (;;) {
@@ -1245,9 +1062,6 @@ public class DistributionManager implements DM {
}
}
- synchronized (DistributionManager.class) {
- openDMs++;
- }
finishedConstructor = true;
} finally {
if (!finishedConstructor) {
@@ -1256,8 +1070,53 @@ public class DistributionManager implements DM {
}
}
+ /**
+ * Is this VM dedicated to administration (like a GUI console or a JMX agent)? If so, then it
+ * creates {@link #ADMIN_ONLY_DM_TYPE} type distribution managers.
+ *
+ * @since GemFire 4.0
+ */
+ public static boolean isDedicatedAdminVM() {
+ return isDedicatedAdminVM;
+ }
+
+ public static void setIsDedicatedAdminVM(boolean isDedicatedAdminVM) {
+ DistributionManager.isDedicatedAdminVM = isDedicatedAdminVM;
+ }
+
+ public static Boolean getIsStartupThread() {
+ return isStartupThread.get();
+ }
+
+ public static void setIsStartupThread(Boolean isStartup) {
+ DistributionManager.isStartupThread.set(isStartup);
+ }
+
//////////////////// Instance Methods /////////////////////
+ private void runUntilShutdown(Runnable r) {
+ try {
+ r.run();
+ } catch (CancelException e) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Caught shutdown exception", e);
+ }
+ } catch (VirtualMachineError err) {
+ SystemFailure.initiateFailure(err);
+ // If this ever returns, rethrow the error. We're poisoned
+ // now, so don't let this thread continue.
+ throw err;
+ } catch (Throwable t) {
+ SystemFailure.checkFailure();
+ if (isCloseInProgress()) {
+ logger.debug("Caught unusual exception during shutdown: {}", t.getMessage(), t);
+ } else {
+ logger.warn(LocalizedMessage
+ .create(LocalizedStrings.DistributionManager_TASK_FAILED_WITH_EXCEPTION), t);
+ }
+ }
+ }
+
/**
* Returns true if the two members are on the same equivalent host based on overlapping IP
* addresses collected for all NICs during exchange of startup messages.
@@ -1265,6 +1124,7 @@ public class DistributionManager implements DM {
* @param member1 First member
* @param member2 Second member
*/
+ @Override
public boolean areOnEquivalentHost(InternalDistributedMember member1,
InternalDistributedMember member2) {
Set<InetAddress> equivalents1 = getEquivalents(member1.getInetAddress());
@@ -1277,7 +1137,7 @@ public class DistributionManager implements DM {
*
* @param equivs list of InetAddress's that all point at same host
*/
- public void setEquivalentHosts(Set<InetAddress> equivs) {
+ void setEquivalentHosts(Set<InetAddress> equivs) {
Iterator<InetAddress> it = equivs.iterator();
synchronized (equivalentHosts) {
while (it.hasNext()) {
@@ -1286,11 +1146,6 @@ public class DistributionManager implements DM {
}
}
- public HashMap<InetAddress, Set<InetAddress>> getEquivalentHostsSnapshot() {
- synchronized (this.equivalentHosts) {
- return new HashMap<InetAddress, Set<InetAddress>>(this.equivalentHosts);
- }
- }
/**
* Return all of the InetAddress's that are equivalent to the given one (same host)
@@ -1298,6 +1153,7 @@ public class DistributionManager implements DM {
* @param in host to match up
* @return all the addresses thus equivalent
*/
+ @Override
public Set<InetAddress> getEquivalents(InetAddress in) {
Set<InetAddress> result;
synchronized (equivalentHosts) {
@@ -1311,7 +1167,7 @@ public class DistributionManager implements DM {
return result;
}
- public void setRedundancyZone(InternalDistributedMember member, String redundancyZone) {
+ void setRedundancyZone(InternalDistributedMember member, String redundancyZone) {
if (redundancyZone != null && !redundancyZone.equals("")) {
this.redundancyZones.put(member, redundancyZone);
}
@@ -1328,10 +1184,11 @@ public class DistributionManager implements DM {
* Set the flag indicating that we should enforce unique zones. If we are already enforcing unique
* zones, keep it that way.
*/
- public void setEnforceUniqueZone(boolean enforceUniqueZone) {
+ void setEnforceUniqueZone(boolean enforceUniqueZone) {
this.enforceUniqueZone |= enforceUniqueZone;
}
+ @Override
public boolean enforceUniqueZone() {
return enforceUniqueZone;
}
@@ -1371,21 +1228,11 @@ public class DistributionManager implements DM {
}
public List<InternalDistributedMember> getViewMembers() {
- NetView result = null;
- DistributionChannel ch = this.channel;
- if (ch != null) {
- MembershipManager mgr = ch.getMembershipManager();
- if (mgr != null) {
- result = mgr.getView();
- }
- }
- if (result == null) {
- result = new NetView();
- }
- return result.getMembers();
+ return membershipManager.getView().getMembers();
}
/* implementation of DM.getOldestMember */
+ @Override
public DistributedMember getOldestMember(Collection c) throws NoSuchElementException {
List<InternalDistributedMember> view = getViewMembers();
for (int i = 0; i < view.size(); i++) {
@@ -1408,22 +1255,9 @@ public class DistributionManager implements DM {
}
/**
- * Print a membership view (list of {@link InternalDistributedMember}s)
- *
- * @param v the list
- * @return String
- */
- public static String printView(NetView v) {
- if (v == null)
- return "null";
-
- return v.toString();
- }
-
- /**
* Need to do this outside the constructor so that the child constructor can finish.
*/
- protected void startThreads() {
+ private void startThreads() {
this.system.setDM(this); // fix for bug 33362
if (this.memberEventThread != null)
this.memberEventThread.start();
@@ -1432,12 +1266,12 @@ public class DistributionManager implements DM {
// And the distinguished guests today are...
NetView v = membershipManager.getView();
logger.info(LocalizedMessage.create(
- LocalizedStrings.DistributionManager_INITIAL_MEMBERSHIPMANAGER_VIEW___0, printView(v)));
+ LocalizedStrings.DistributionManager_INITIAL_MEMBERSHIPMANAGER_VIEW___0,
+ String.valueOf(v)));
// Add them all to our view
- Iterator<InternalDistributedMember> it = v.getMembers().iterator();
- while (it.hasNext()) {
- addNewMember(it.next());
+ for (InternalDistributedMember internalDistributedMember : v.getMembers()) {
+ addNewMember(internalDistributedMember);
}
// Figure out who the elder is...
@@ -1452,7 +1286,7 @@ public class DistributionManager implements DM {
public void run() {
// call in background since it might need to send a reply
// and we are not ready to send messages until startup is finished
- isStartupThread.set(Boolean.TRUE);
+ setIsStartupThread(Boolean.TRUE);
readyForMessages();
}
});
@@ -1473,7 +1307,7 @@ public class DistributionManager implements DM {
}
}
- protected void readyForMessages() {
+ private void readyForMessages() {
synchronized (this) {
this.readyForMessages = true;
this.notifyAll();
@@ -1481,10 +1315,9 @@ public class DistributionManager implements DM {
membershipManager.startEventProcessing();
}
- protected void waitUntilReadyForMessages() {
+ private void waitUntilReadyForMessages() {
if (readyForMessages)
return;
- // membershipManager.waitForEventProcessing();
synchronized (this) {
for (;;) {
if (readyForMessages)
@@ -1520,7 +1353,7 @@ public class DistributionManager implements DM {
*
* @param msg the messsage that is currently being sent
*/
- protected void waitUntilReadyToSendMsgs(DistributionMessage msg) {
+ private void waitUntilReadyToSendMsgs(DistributionMessage msg) {
if (this.readyToSendMsgs) {
return;
}
@@ -1530,13 +1363,13 @@ public class DistributionManager implements DM {
|| msg instanceof AdminMessageType) {
return;
}
- if (isStartupThread.get() != null) {
+ if (getIsStartupThread() == Boolean.TRUE) {
// let the startup thread send messages
// the only case I know of that does this is if we happen to log a
// message during startup and an alert listener has registered.
return;
}
- // membershipManager.waitForEventProcessing();
+
synchronized (this.readyToSendMsgsLock) {
for (;;) {
if (this.readyToSendMsgs)
@@ -1557,13 +1390,13 @@ public class DistributionManager implements DM {
} // synchronized
}
- // DM method
+
@Override
public void forceUDPMessagingForCurrentThread() {
membershipManager.forceUDPMessagingForCurrentThread();
}
- // DM method
+
@Override
public void releaseUDPMessagingForCurrentThread() {
membershipManager.releaseUDPMessagingForCurrentThread();
@@ -1589,6 +1422,7 @@ public class DistributionManager implements DM {
* Returns the current "cache time" in milliseconds since the epoch. The "cache time" takes into
* account skew among the local clocks on the various machines involved in the cache.
*/
+ @Override
public long cacheTimeMillis() {
return this.system.getClock().cacheTimeMillis();
}
@@ -1598,14 +1432,16 @@ public class DistributionManager implements DM {
/**
* Returns the id of this distribution manager.
*/
+ @Override
public InternalDistributedMember getDistributionManagerId() {
- return this.myid;
+ return this.localAddress;
}
/**
* Returns an unmodifiable set containing the identities of all of the known (non-admin-only)
* distribution managers.
*/
+ @Override
public Set getDistributionManagerIds() {
// access to members synchronized under membersLock in order to
// ensure serialization
@@ -1621,6 +1457,7 @@ public class DistributionManager implements DM {
*
* @since GemFire 6.6.3
*/
+ @Override
public void addHostedLocators(InternalDistributedMember member, Collection<String> locators,
boolean isSharedConfigurationEnabled) {
synchronized (this.membersLock) {
@@ -1630,11 +1467,6 @@ public class DistributionManager implements DM {
if (this.hostedLocatorsAll.isEmpty()) {
this.hostedLocatorsAll = new HashMap<InternalDistributedMember, Collection<String>>();
}
-
- if (!this.isSharedConfigEnabledForDS) {
- this.isSharedConfigEnabledForDS = isSharedConfigurationEnabled;
- }
-
Map<InternalDistributedMember, Collection<String>> tmp =
new HashMap<InternalDistributedMember, Collection<String>>(this.hostedLocatorsAll);
tmp.remove(member);
@@ -1647,11 +1479,9 @@ public class DistributionManager implements DM {
throw new IllegalArgumentException("Cannot use empty collection of locators");
}
if (this.hostedLocatorsWithSharedConfiguration.isEmpty()) {
- this.hostedLocatorsWithSharedConfiguration =
- new HashMap<InternalDistributedMember, Collection<String>>();
+ this.hostedLocatorsWithSharedConfiguration = new HashMap<>();
}
- tmp = new HashMap<InternalDistributedMember, Collection<String>>(
- this.hostedLocatorsWithSharedConfiguration);
+ tmp = new HashMap<>(this.hostedLocatorsWithSharedConfiguration);
tmp.remove(member);
tmp.put(member, locators);
tmp = Collections.unmodifiableMap(tmp);
@@ -1666,7 +1496,7 @@ public class DistributionManager implements DM {
synchronized (this.membersLock) {
if (this.hostedLocatorsAll.containsKey(member)) {
Map<InternalDistributedMember, Collection<String>> tmp =
- new HashMap<InternalDistributedMember, Collection<String>>(this.hostedLocatorsAll);
+ new HashMap<>(this.hostedLocatorsAll);
tmp.remove(member);
if (tmp.isEmpty()) {
tmp = Collections.emptyMap();
@@ -1699,6 +1529,7 @@ public class DistributionManager implements DM {
*
* @since GemFire 6.6.3
*/
+ @Override
public Collection<String> getHostedLocators(InternalDistributedMember member) {
synchronized (this.membersLock) {
return this.hostedLocatorsAll.get(member);
@@ -1744,6 +1575,7 @@ public class DistributionManager implements DM {
* Returns an unmodifiable set containing the identities of all of the known (including admin)
* distribution managers.
*/
+ @Override
public Set getDistributionManagerIdsIncludingAdmin() {
// access to members synchronized under membersLock in order to
// ensure serialization
@@ -1754,20 +1586,10 @@ public class DistributionManager implements DM {
/**
- * Returns the low-level distribution channel for this distribution manager. (brought over from
- * ConsoleDistributionManager)
- *
- * @since GemFire 4.0
- */
- public DistributionChannel getDistributionChannel() {
- return this.channel;
- }
-
-
- /**
* Returns a private-memory list containing the identities of all the other known distribution
* managers not including me.
*/
+ @Override
public Set getOtherDistributionManagerIds() {
// We return a modified copy of the list, so
// collect the old list and copy under the lock.
@@ -1793,6 +1615,7 @@ public class DistributionManager implements DM {
return result;
}
+ @Override
public InternalDistributedMember getCanonicalId(DistributedMember id) {
// the members set is copy-on-write, so it is safe to iterate over it
InternalDistributedMember result = this.members.get(id);
@@ -1805,6 +1628,7 @@ public class DistributionManager implements DM {
/**
* Add a membership listener and return other DistributionManagerIds as an atomic operation
*/
+ @Override
public Set addMembershipListenerAndGetDistributionManagerIds(MembershipListener l) {
// switched sync order to fix bug 30360
synchronized (this.membersLock) {
@@ -1818,7 +1642,7 @@ public class DistributionManager implements DM {
}
}
- public void addNewMember(InternalDistributedMember member) {
+ private void addNewMember(InternalDistributedMember member) {
// This is the place to cleanup the zombieMembers
int vmType = member.getVmKind();
switch (vmType) {
@@ -1838,38 +1662,17 @@ public class DistributionManager implements DM {
/**
* Returns the identity of this <code>DistributionManager</code>
*/
+ @Override
public InternalDistributedMember getId() {
- return this.myid;
+ return this.localAddress;
}
- /**
- * Returns the id of the underlying distribution channel used for communication.
- *
- * @since GemFire 3.0
- */
- public long getChannelId() {
- return this.channel.getId();
- }
-
- /**
- * Adds a message to the outgoing queue. Note that <code>message</code> should not be modified
- * after it has been added to the queue. After <code>message</code> is distributed, it will be
- * recycled.
- *
- * @return list of recipients who did not receive the message
- * @throws NotSerializableException if the content is not serializable
- */
- public Set putOutgoingUserData(final DistributionMessage message)
- throws NotSerializableException {
- return sendMessage(message);
+ @Override
+ public long getMembershipPort() {
+ return localAddress.getPort();
}
- /**
- * Send outgoing data; message is guaranteed to be serialized.
- *
- * @return list of recipients who did not receive the message
- * @throws InternalGemFireException if message is not serializable
- */
+ @Override
public Set putOutgoing(final DistributionMessage msg) {
try {
DistributionMessageObserver observer = DistributionMessageObserver.getInstance();
@@ -1891,11 +1694,6 @@ public class DistributionManager implements DM {
}
/**
- * @see #closeInProgress
- */
- private final Object shutdownMutex = new Object();
-
- /**
* Informs other members that this dm is shutting down. Stops the pusher, puller, and processor
* threads and closes the connection to the transport layer.
*/
@@ -1915,7 +1713,7 @@ public class DistributionManager implements DM {
: "");
logger.info(LocalizedMessage.create(
LocalizedStrings.DistributionManager_SHUTTING_DOWN_DISTRIBUTIONMANAGER_0_1,
- new Object[] {this.myid, exceptionStatus}));
+ new Object[] {this.localAddress, exceptionStatus}));
final long start = System.currentTimeMillis();
try {
@@ -1940,7 +1738,7 @@ public class DistributionManager implements DM {
};
final Thread t = new Thread(threadGroup, r,
LocalizedStrings.DistributionManager_SHUTDOWN_MESSAGE_THREAD_FOR_0
- .toLocalizedString(this.myid));
+ .toLocalizedString(this.localAddress));
t.start();
boolean interrupted = Thread.interrupted();
try {
@@ -2066,21 +1864,6 @@ public class DistributionManager implements DM {
}
/**
- * maximum time, in milliseconds, to wait for all threads to exit
- */
- private static final int MAX_STOP_TIME = 20000;
-
- /**
- * Time to sleep, in milliseconds, while polling to see if threads have finished
- */
- private static final int STOP_PAUSE_TIME = 1000;
-
- /**
- * Maximum number of interrupt attempts to stop a thread
- */
- private static final int MAX_STOP_ATTEMPTS = 10;
-
- /**
* Cheap tool to kill a referenced thread
*
* @param t the thread to kill
@@ -2250,6 +2033,7 @@ public class DistributionManager implements DM {
/** the latest view ID that has been processed by all membership listeners */
private long membershipViewIdAcknowledged;
+ @Override
public boolean shutdownInProgress() {
return this.shutdownInProgress;
}
@@ -2263,8 +2047,8 @@ public class DistributionManager implements DM {
this.closeInProgress = true; // set here also to fix bug 36736
removeAllHealthMonitors();
shutdownInProgress = true;
- if (this.channel != null) {
- this.channel.setShutDown();
+ if (membershipManager != null) {
+ membershipManager.setShutdown();
}
askThreadsToStop();
@@ -2277,12 +2061,6 @@ public class DistributionManager implements DM {
}
forceThreadsToStop();
- // // bug36329: desperation measure, send a second interrupt?
- // try { Thread.sleep(1000); }
- // catch (InterruptedException ie) {
- // // No need to reset interrupt bit, we're really trying to quit...
- // }
- // forceThreadsToStop();
} // try
finally {
// ABSOLUTELY ESSENTIAL that we close the distribution channel!
@@ -2297,55 +2075,43 @@ public class DistributionManager implements DM {
}
}
} finally {
- if (this.channel != null) {
+ if (this.membershipManager != null) {
logger.info(LocalizedMessage.create(
- LocalizedStrings.DistributionManager_NOW_CLOSING_DISTRIBUTION_FOR__0, this.myid));
- this.channel.disconnect(beforeJoined);
- // this.channel = null; DO NOT NULL OUT INSTANCE VARIABLES AT SHUTDOWN - bug #42087
+ LocalizedStrings.DistributionManager_NOW_CLOSING_DISTRIBUTION_FOR__0,
+ this.localAddress));
+ this.membershipManager.disconnect(beforeJoined);
}
}
}
}
- /**
- * Returns the distributed system to which this distribution manager is connected.
- */
+ @Override
public InternalDistributedSystem getSystem() {
return this.system;
}
/**
* Returns the transport configuration for this distribution manager
- *
- * @since GemFire 5.0
*/
- public RemoteTransportConfig getTransport() {
+ RemoteTransportConfig getTransport() {
return this.transport;
}
- /**
- * Adds a <code>MembershipListener</code> to this distribution manager.
- */
+ @Override
public void addMembershipListener(MembershipListener l) {
this.membershipListeners.putIfAbsent(l, Boolean.TRUE);
}
- /**
- * Removes a <code>MembershipListener</code> from this distribution manager.
- *
- * @throws IllegalArgumentException <code>l</code> was not registered on this distribution manager
- */
+ @Override
public void removeMembershipListener(MembershipListener l) {
this.membershipListeners.remove(l);
}
/**
* Adds a <code>MembershipListener</code> to this distribution manager.
- *
- * @since GemFire 5.7
*/
- public void addAllMembershipListener(MembershipListener l) {
+ private void addAllMembershipListener(MembershipListener l) {
synchronized (this.allMembershipListenersLock) {
Set newAllMembershipListeners = new HashSet(this.allMembershipListeners);
newAllMembershipListeners.add(l);
@@ -2353,13 +2119,7 @@ public class DistributionManager implements DM {
}
}
- /**
- * Removes a <code>MembershipListener</code> listening for all members from this distribution
- * manager.
- *
- * @throws IllegalArgumentException <code>l</code> was not registered on this distribution manager
- * @since GemFire 5.7
- */
+ @Override
public void removeAllMembershipListener(MembershipListener l) {
synchronized (this.allMembershipListenersLock) {
Set newAllMembershipListeners = new HashSet(this.allMembershipListeners);
@@ -2404,7 +2164,7 @@ public class DistributionManager implements DM {
* This stalls waiting for the current membership view (as seen by the membership manager) to be
* acknowledged by all membership listeners
*/
- public void waitForViewInstallation(long id) throws InterruptedException {
+ void waitForViewInstallation(long id) throws InterruptedException {
if (id <= this.membershipViewIdAcknowledged) {
return;
}
@@ -2419,7 +2179,7 @@ public class DistributionManager implements DM {
}
}
- protected void handleMemberEvent(MemberEvent ev) {
+ private void handleMemberEvent(MemberEvent ev) {
ev.handleEvent(this);
}
@@ -2506,23 +2266,19 @@ public class DistributionManager implements DM {
}
- /**
- * Stops the threads associated with this distribution manager and closes the connection to the
- * transport layer.
- */
+ @Override
public void close() {
if (!closed) {
this.shutdown();
logger.info(LocalizedMessage.create(
- LocalizedStrings.DistributionManager_MARKING_DISTRIBUTIONMANAGER_0_AS_CLOSED, this.myid));
- MembershipLogger.logShutdown(this.myid);
+ LocalizedStrings.DistributionManager_MARKING_DISTRIBUTIONMANAGER_0_AS_CLOSED,
+ this.localAddress));
+ MembershipLogger.logShutdown(this.localAddress);
closed = true;
- synchronized (DistributionManager.class) {
- openDMs--;
- }
}
}
+ @Override
public void throwIfDistributionStopped() {
if (this.shutdownMsgSent) {
throw new DistributedSystemDisconnectedException(
@@ -2539,9 +2295,8 @@ public class DistributionManager implements DM {
return this.closed;
}
- /**
- * Makes note of a new administration console (admin-only member).
- */
+
+ @Override
public void addAdminConsole(InternalDistributedMember theId) {
logger.info(LocalizedMessage.create(
LocalizedStrings.DistributionManager_NEW_ADMINISTRATION_MEMBER_DETECTED_AT_0, theId));
@@ -2552,10 +2307,12 @@ public class DistributionManager implements DM {
}
}
+ @Override
public DMStats getStats() {
return this.stats;
}
+ @Override
public DistributionConfig getConfig() {
DistributionConfig result = null;
InternalDistributedSystem sys = getSystem();
@@ -2565,32 +2322,14 @@ public class DistributionManager implements DM {
return result;
}
- // /**
- // * Initializes and returns a <code>DistributedSystem</code> to be
- // * sent to new members of the distributed system.
- // *
- // * @since GemFire 3.0
- // */
- // protected DistributedState getNewDistributedState() {
- // DistributedState state = new DistributedState();
- // state.setGemFireVersion(GemFireVersion.getGemFireVersion());
- // state.setCacheTime(this.cacheTimeMillis());
- // return state;
- // }
-
- private static final int STARTUP_TIMEOUT =
- Integer.getInteger("DistributionManager.STARTUP_TIMEOUT", 15000).intValue();
-
- public static final boolean DEBUG_NO_ACKNOWLEDGEMENTS =
- Boolean.getBoolean("DistributionManager.DEBUG_NO_ACKNOWLEDGEMENTS");
-
+ @Override
public Set getAllOtherMembers() {
Set result = new HashSet(getDistributionManagerIdsIncludingAdmin());
result.remove(getDistributionManagerId());
return result;
}
- @Override // DM method
+ @Override
public void retainMembersWithSameOrNewerVersion(Collection<InternalDistributedMember> members,
Version version) {
for (Iterator<InternalDistributedMember> it = members.iterator(); it.hasNext();) {
@@ -2601,7 +2340,7 @@ public class DistributionManager implements DM {
}
}
- @Override // DM method
+ @Override
public void removeMembersWithSameOrNewerVersion(Collection<InternalDistributedMember> members,
Version version) {
for (Iterator<InternalDistributedMember> it = members.iterator(); it.hasNext();) {
@@ -2612,48 +2351,20 @@ public class DistributionManager implements DM {
}
}
- /**
- * Add a membership listener for all members and return other DistribtionManagerIds as an atomic
- * operation
- *
- * @since GemFire 5.7
- */
+ @Override
public Set addAllMembershipListenerAndGetAllIds(MembershipListener l) {
- // TO fix this deadlock:
- // "View Message Processor":
- // waiting to lock monitor 0x080f691c (object 0xe3ba7680, a
- // org.apache.geode.distributed.internal.DistributionManager$MembersLock),
- // which is held by "RMI TCP Connection(259)-10.80.10.55"
- // "RMI TCP Connection(259)-10.80.10.55":
- // waiting to lock monitor 0x080f6598 (object 0xe3bacd90, a
- // org.apache.geode.distributed.internal.membership.jgroup.JGroupMembershipManager$ViewLock),
- // which is held by "View Message Processor"
- // NEED to prevent view changes while installing a listener.
- DistributionChannel ch = this.channel;
- if (ch != null) {
- MembershipManager mgr = ch.getMembershipManager();
- if (mgr != null) {
- mgr.getViewLock().writeLock().lock();
- try {
- synchronized (this.membersLock) {
- // Don't let the members come and go while we are adding this
- // listener. This ensures that the listener (probably a
- // ReplyProcessor) gets a consistent view of the members.
- addAllMembershipListener(l);
- return getDistributionManagerIdsIncludingAdmin();
- }
- } finally {
- mgr.getViewLock().writeLock().unlock();
- }
+ MembershipManager mgr = membershipManager;
+ mgr.getViewLock().writeLock().lock();
+ try {
+ synchronized (this.membersLock) {
+ // Don't let the members come and go while we are adding this
+ // listener. This ensures that the listener (probably a
+ // ReplyProcessor) gets a consistent view of the members.
+ addAllMembershipListener(l);
+ return getDistributionManagerIdsIncludingAdmin();
}
- }
- // If we have no channel or MembershipManager then the view is empty
- synchronized (this.membersLock) {
- // Don't let the members come and go while we are adding this
- // listener. This ensures that the listener (probably a
- // ReplyProcessor) gets a consistent view of the members.
- addAllMembershipListener(l);
- return Collections.EMPTY_SET;
+ } finally {
+ mgr.getViewLock().writeLock().unlock();
}
}
@@ -2661,7 +2372,7 @@ public class DistributionManager implements DM {
* Sends a startup message and waits for a response. Returns true if response received; false if
* it timed out or there are no peers.
*/
- protected boolean sendStartupMessage(StartupOperation startupOperation, boolean cancelOnTimeout)
+ private boolean sendStartupMessage(StartupOperation startupOperation)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
@@ -2727,7 +2438,7 @@ public class DistributionManager implements DM {
boolean isAdminDM = getId().getVmKind() == DistributionManager.ADMIN_ONLY_DM_TYPE
|| getId().getVmKind() == DistributionManager.LOCATOR_DM_TYPE
- || DistributionManager.isDedicatedAdminVM
+ || DistributionManager.isDedicatedAdminVM()
|| Boolean.getBoolean(InternalLocator.FORCE_LOCATOR_DM_TYPE);
boolean receivedAny = this.receivedStartupResponse;
@@ -2806,6 +2517,7 @@ public class DistributionManager implements DM {
*/
private final Object unfinishedStartupsLock = new Object();
+ @Override
public void setUnfinishedStartups(Collection s) {
synchronized (unfinishedStartupsLock) {
Assert.assertTrue(unfinishedStartups == null, "Set unfinished startups twice");
@@ -2825,6 +2537,7 @@ public class DistributionManager implements DM {
}
}
+ @Override
public void removeUnfinishedStartup(InternalDistributedMember m, boolean departed) {
synchronized (unfinishedStartupsLock) {
if (logger.isDebugEnabled()) {
@@ -2857,25 +2570,6 @@ public class DistributionManager implements DM {
*
* @see StartupResponseMessage#process
*/
- void processStartupResponse(InternalDistributedMember sender, long otherCacheTime,
- String theRejectionMessage) {
- removeUnfinishedStartup(sender, false);
- synchronized (this) {
- if (!this.receivedStartupResponse) {
- this.receivedStartupResponse = true;
- }
- if (theRejectionMessage != null && this.rejectionMessage == null) {
- // remember the first non-null rejection. This fixes bug 33266
- this.rejectionMessage = theRejectionMessage;
- }
- }
- }
-
- /**
- * Processes the first startup response.
- *
- * @see StartupResponseMessage#process
- */
void processStartupResponse(InternalDistributedMember sender, String theRejectionMessage) {
removeUnfinishedStartup(sender, false);
synchronized (this) {
@@ -2899,7 +2593,7 @@ public class DistributionManager implements DM {
List<InternalDistributedMember> theMembers = getViewMembers();
// Assert.assertTrue(!closeInProgress
- // && theMembers.contains(this.myid)); // bug36202?
+ // && theMembers.contains(this.localAddress)); // bug36202?
int elderCandidates = 0;
Iterator<InternalDistributedMember> it;
@@ -2967,14 +2661,15 @@ public class DistributionManager implements DM {
continue;
}
- if (member.equals(this.myid)) {
+ if (member.equals(this.localAddress)) {
if (!this.adam && elderCandidates == 1) {
this.adam = true;
logger.info(LocalizedMessage.create(
- LocalizedStrings.DistributionManager_0_IS_THE_ELDER_AND_THE_ONLY_MEMBER, this.myid));
+ LocalizedStrings.DistributionManager_0_IS_THE_ELDER_AND_THE_ONLY_MEMBER,
+ this.localAddress));
} else {
logger.info(LocalizedMessage.create(LocalizedStrings.DistributionManager_I_0_AM_THE_ELDER,
- this.myid));
+ this.localAddress));
}
}
return member;
@@ -2987,12 +2682,12 @@ public class DistributionManager implements DM {
* Select a new elder
*
*/
- protected void selectElder() {
+ private void selectElder() {
getSystem().getCancelCriterion().checkCancelInProgress(null); // bug 37884, if DS is
// disconnecting, throw exception
// Once we are the elder, we're stuck until we leave the view.
- if (this.myid.equals(this.elder)) {
+ if (this.localAddress.equals(this.elder)) {
return;
}
@@ -3072,7 +2767,7 @@ public class DistributionManager implements DM {
} // if
// In any event, make sure that this member is no longer an elder.
- if (!theId.equals(myid) && theId.equals(elder)) {
+ if (!theId.equals(localAddress) && theId.equals(elder)) {
try {
selectElder();
} catch (DistributedSystemDisconnectedException e) {
@@ -3120,9 +2815,7 @@ public class DistributionManager implements DM {
addMemberEvent(new MemberJoinedEvent(theId));
}
- /**
- * Return true if id is a current member of our system.
- */
+ @Override
public boolean isCurrentMember(InternalDistributedMember id) {
Set m;
synchronized (this.membersLock) {
@@ -3167,21 +2860,11 @@ public class DistributionManager implements DM {
* Process an incoming distribution message. This includes scheduling it correctly based on the
* message's nioPriority (executor type)
*/
- public void handleIncomingDMsg(DistributionMessage message) {
- /*
- * disabled - not being used if (message instanceof OutgoingMessageWrapper) {
- * putOutgoing(((OutgoingMessageWrapper)message).getMessage()); return; }
- */
-
- // long latency = message.getLatency();
- // this.stats.incMessageTransitTime(latency * 1000000L);
- // message.resetTimestamp();
+ private void handleIncomingDMsg(DistributionMessage message) {
stats.incReceivedMessages(1L);
stats.incReceivedBytes(message.getBytesRead());
stats.incMessageChannelTime(message.resetTimestamp());
- // message.setRecipient(DistributionManager.this.getId());
-
if (logger.isDebugEnabled()) {
logger.debug("Received message '{}' from <{}>", message, message.getSender());
}
@@ -3266,11 +2949,7 @@ public class DistributionManager implements DM {
LocalizedStrings.ShutdownMessage_SHUTDOWN_MESSAGE_RECEIVED.toLocalizedString());
}
- /**
- * used by the DistributedMembershipListener and startup and shutdown operations, this method
- * decrements the number of nodes and handles lower-level clean up of the resources used by the
- * departed manager
- */
+ @Override
public void handleManagerDeparture(InternalDistributedMember theId, boolean p_crashed,
String p_reason) {
boolean crashed = p_crashed;
@@ -3325,7 +3004,7 @@ public class DistributionManager implements DM {
}
}
- public void handleManagerSuspect(InternalDistributedMember suspect,
+ private void handleManagerSuspect(InternalDistributedMember suspect,
InternalDistributedMember whoSuspected, String reason) {
if (!isCurrentMember(suspect)) {
return; // fault tolerance
@@ -3339,11 +3018,11 @@ public class DistributionManager implements DM {
addMemberEvent(new MemberSuspectEvent(suspect, whoSuspected, reason));
}
- public void handleViewInstalled(NetView view) {
+ private void handleViewInstalled(NetView view) {
addMemberEvent(new ViewInstalledEvent(view));
}
- public void handleQuorumLost(Set<InternalDistributedMember> failures,
+ private void handleQuorumLost(Set<InternalDistributedMember> failures,
List<InternalDistributedMember> remaining) {
addMemberEvent(new QuorumLostEvent(failures, remaining));
}
@@ -3351,7 +3030,7 @@ public class DistributionManager implements DM {
/**
* Sends the shutdown message. Not all DistributionManagers need to do this.
*/
- protected void sendShutdownMessage() {
+ private void sendShutdownMessage() {
if (getDMType() == ADMIN_ONLY_DM_TYPE && Locator.getLocators().size() == 0) {
// [bruce] changed above "if" to have ShutdownMessage sent by locators.
// Otherwise the system can hang because an admin member does not trigger
@@ -3365,9 +3044,6 @@ public class DistributionManager implements DM {
m.setDistributionManagerId(theId);
Set allOthers = new HashSet(getViewMembers());
allOthers.remove(getDistributionManagerId());
- // ReplyProcessor21 rp = new ReplyProcessor21(this, allOthers);
- // m.setProcessorId(rp.getProcessorId());
- // m.setMulticast(system.getConfig().getMcastPort() != 0);
m.setRecipients(allOthers);
// Address recipient = (Address) m.getRecipient();
@@ -3379,7 +3055,7 @@ public class DistributionManager implements DM {
try {
// m.resetTimestamp(); // nanotimers across systems don't match
long startTime = DistributionStats.getStatTime();
- channel.send(m.getRecipients(), m, this, stats);
+ sendViaMembershipManager(m.getRecipients(), m, this, stats);
this.stats.incSentMessages(1L);
if (DistributionStats.enableClockStats) {
stats.incSentMessagesTime(DistributionStats.getStatTime() - startTime);
@@ -3421,36 +3097,6 @@ public class DistributionManager implements DM {
}
}
- // /**
- // * Return a shortened name of a class that excludes the package
- // */
- // private static String shortenClassName(String className) {
- // int index = className.lastIndexOf('.');
- // if (index != -1) {
- // return className.substring(index + 1);
- //
- // } else {
- // return className;
- // }
- // }
-
- /**
- * Send a message that is guaranteed to be serialized
- *
- * @param msg
- * @return the recipients who did not receive the message
- */
- protected Set sendOutgoingSerialized(DistributionMessage msg) {
- try {
- return sendOutgoing(msg);
- } catch (NotSerializableException e) {
- throw new InternalGemFireException(e);
- } catch (ToDataException e) {
- // exception from user code
- throw e;
- }
- }
-
/**
* Actually does the work of sending a message out over the distribution channel.
*
@@ -3459,11 +3105,11 @@ public class DistributionManager implements DM {
* all received it or it was sent to {@link DistributionMessage#ALL_RECIPIENTS}.
* @throws NotSerializableException If <code>message</code> cannot be serialized
*/
- protected Set sendOutgoing(DistributionMessage message) throws NotSerializableException {
+ Set sendOutgoing(DistributionMessage message) throws NotSerializableException {
long startTime = DistributionStats.getStatTime();
- Set result =
- channel.send(message.getRecipients(), message, DistributionManager.this, this.stats);
+ Set result = sendViaMembershipManager(message.getRecipients(), message,
+ DistributionManager.this, this.stats);
long endTime = 0L;
if (DistributionStats.enableClockStats) {
endTime = NanoTimer.getTime();
@@ -3501,20 +3147,9 @@ public class DistributionManager implements DM {
waitUntilReadyToSendMsgs(message);
result = sendOutgoing(message);
- } catch (NotSerializableException ex) {
- throw ex; // serialization error in user data
- } catch (ToDataException ex) {
- throw ex; // serialization error in user data
- } catch (ReenteredConnectException ex) {
- throw ex; // Recursively tried to get the same connection
- } catch (CancelException ex) {
- throw ex; // bug 37194, shutdown conditions
- } catch (InvalidDeltaException ide) {
- logger.info(
- LocalizedMessage
- .create(LocalizedStrings.DistributionManager_CAUGHT_EXCEPTION_WHILE_SENDING_DELTA),
- ide.getCause());
- throw (RuntimeException) ide.getCause();
+ } catch (NotSerializableException | ToDataException | ReenteredConnectException
+ | InvalidDeltaException | CancelException ex) {
+ throw ex;
} catch (Exception ex) {
DistributionManager.this.exceptionInThreads = true;
String receiver = "NULL";
@@ -3539,13 +3174,33 @@ public class DistributionManager implements DM {
return result;
}
+ /**
+ * @return list of recipients who did not receive the message because they left the view (null if
+ * all received it or it was sent to {@link DistributionMessage#ALL_RECIPIENTS}).
+ * @throws NotSerializableException If content cannot be serialized
+ */
+ private Set sendViaMembershipManager(InternalDistributedMember[] destinations,
+ DistributionMessage content, DistributionManager dm, DistributionStats stats)
+ throws NotSerializableException {
+ if (membershipManager == null) {
+ logger.warn(LocalizedMessage.create(
+ LocalizedStrings.DistributionChannel_ATTEMPTING_A_SEND_TO_A_DISCONNECTED_DISTRIBUTIONMANAGER));
+ if (destinations.length == 1 && destinations[0] == DistributionMessage.ALL_RECIPIENTS)
+ return null;
+ HashSet result = new HashSet();
+ for (int i = 0; i < destinations.length; i++) {
+ result.add(destinations[i]);
+ }
+ return result;
+ }
+ return membershipManager.send(destinations, content, stats);
+ }
+
/**
* Schedule a given message appropriately, depending upon its executor kind.
- *
- * @param message
*/
- protected void scheduleIncomingMessage(DistributionMessage message) {
+ void scheduleIncomingMessage(DistributionMessage message) {
/*
* Potential race condition between starting up and getting other distribution manager ids -- DM
* will only be initialized upto the point at which it called startThreads
@@ -3554,39 +3209,13 @@ public class DistributionManager implements DM {
message.schedule(DistributionManager.this);
}
- /**
- * Mutex to control access to {@link #waitingForElderChange} or {@link #elder}.
- */
- protected final Object elderMonitor = new Object();
-
- /**
- * Must be read/written while holding {@link #elderMonitor}
- *
- * @see #elderChangeWait()
- */
- private boolean waitingForElderChange = false;
-
- /**
- * @see DM#isAdam()
- */
- private boolean adam = false;
-
- /**
- * This is the "elder" member of the distributed system, responsible for certain types of
- * arbitration.
- *
- * Must hold {@link #elderMonitor} in order to change this.
- *
- * @see #getElderId()
- */
- protected volatile InternalDistributedMember elder = null;
-
+ @Override
public boolean isAdam() {
return this.adam;
}
+ @Override
public InternalDistributedMember getElderId() throws DistributedSystemDisconnectedException {
- // membershipManager.waitForEventProcessing();
if (closeInProgress) {
throw new DistributedSystemDisconnectedException(
LocalizedStrings.DistributionManager_NO_VALID_ELDER_WHEN_SYSTEM_IS_SHUTTING_DOWN
@@ -3610,10 +3239,12 @@ public class DistributionManager implements DM {
return elder;
}
+ @Override
public boolean isElder() {
return getId().equals(elder);
}
+ @Override
public boolean isLoner() {
return false;
}
@@ -3622,16 +3253,17 @@ public class DistributionManager implements DM {
private ElderState elderState;
private volatile boolean elderStateInitialized;
+ @Override
public ElderState getElderState(boolean force, boolean useTryLock) {
if (force) {
if (logger.isDebugEnabled()) {
- if (!this.myid.equals(this.elder)) {
- logger.debug("Forcing myself, {}, to be the elder.", this.myid);
+ if (!this.localAddress.equals(this.elder)) {
+ logger.debug("Forcing myself, {}, to be the elder.", this.localAddress);
}
}
- changeElder(this.myid);
+ changeElder(this.localAddress);
}
- if (force || this.myid.equals(elder)) {
+ if (force || this.localAddress.equals(elder)) {
// we are the elder
if (this.elderStateInitialized) {
return this.elderState;
@@ -3720,7 +3352,7 @@ public class DistributionManager implements DM {
if (!isCurrentMember(desiredElder)) {
return false; // no longer present
}
- if (this.myid.equals(currentElder)) {
+ if (this.localAddress.equals(currentElder)) {
// Once we become the elder we no longer allow anyone else to be the
// elder so don't let them wait anymore.
return false;
@@ -3763,10 +3395,10 @@ public class DistributionManager implements DM {
/**
* Set the elder to newElder and notify anyone waiting for it to change
*/
- protected void changeElder(InternalDistributedMember newElder) {
+ private void changeElder(InternalDistributedMember newElder) {
synchronized (this.elderMonitor) {
- if (newElder != null && this.myid != null && !this.myid.equals(newElder)) {
- if (this.myid.equals(this.elder)) {
+ if (newElder != null && this.localAddress != null && !this.localAddress.equals(newElder)) {
+ if (this.localAddress.equals(this.elder)) {
// someone else changed the elder while this thread was off cpu
if (logger.isDebugEnabled()) {
logger.debug("changeElder found this VM to be the elder and is taking an early out");
@@ -3785,7 +3417,7 @@ public class DistributionManager implements DM {
/**
* Used to wakeup someone in elderChangeWait even though the elder has not changed
*/
- protected void notifyElderChangeWaiters() {
+ private void notifyElderChangeWaiters() {
synchronized (this.elderMonitor) {
if (this.waitingForElderChange) {
this.waitingForElderChange = false;
@@ -3818,38 +3450,27 @@ public class DistributionManager implements DM {
} // while
}
- /**
- * getThreadPool gets this distribution manager's message-processing thread pool
- */
+ @Override
public ExecutorService getThreadPool() {
return this.threadPool;
}
- /**
- * Return the high-priority message-processing executor
- */
+ @Override
public ExecutorService getHighPriorityThreadPool() {
return this.highPriorityPool;
}
- /**
- * Return the waiting message-processing executor
- */
+ @Override
public ExecutorService getWaitingThreadPool() {
return this.waitingPool;
}
- /**
- * Return the waiting message-processing executor
- */
+ @Override
public ExecutorService getPrMetaDataCleanupThreadPool() {
return this.prMetaDataCleanupThreadPool;
}
- /**
- * Return the waiting message-processing executor
- */
- public Executor getPartitionedRegionExcecutor() {
+ private Executor getPartitionedRegionExcecutor() {
if (this.partitionedRegionThread != null) {
return this.partitionedRegionThread;
} else {
@@ -3857,9 +3478,8 @@ public class DistributionManager implements DM {
}
}
- /**
- * Return the waiting message-processing executor
- */
+
+ @Override
public Executor getFunctionExcecutor() {
if (this.functionExecutionThread != null) {
return this.functionExecutionThread;
@@ -3887,8 +3507,6 @@ public class DistributionManager implements DM {
/**
* Sets the administration agent associated with this distribution manager.
- *
- * @since GemFire 4.0
*/
public void setAgent(RemoteGfManagerAgent agent) {
// Don't let the agent be set twice. There should be a one-to-one
@@ -3912,8 +3530,6 @@ public class DistributionManager implements DM {
/**
* Returns the agent that owns this distribution manager. (in ConsoleDistributionManager)
- *
- * @since GemFire 3.5
*/
public RemoteGfManagerAgent getAgent() {
return this.agent;
@@ -3925,8 +3541,6 @@ public class DistributionManager implements DM {
*
* @return <code>null</code> if no admin {@linkplain #getAgent agent} is associated with this
* distribution manager
- *
- * @since GemFire 3.5
*/
public String getDistributionConfigDescription() {
if (this.agent == null) {
@@ -3950,6 +3564,7 @@ public class DistributionManager implements DM {
* a monitor.
* @since GemFire 3.5
*/
+ @Override
public HealthMonitor getHealthMonitor(InternalDistributedMember owner) {
return (HealthMonitor) this.hmMap.get(owner);
}
@@ -3961,6 +3576,7 @@ public class DistributionManager implements DM {
* @param cfg the configuration to use when creating the monitor
* @since GemFire 3.5
*/
+ @Override
public void createHealthMonitor(InternalDistributedMember owner, GemFireHealthConfig cfg) {
if (closeInProgress) {
return;
@@ -3984,6 +3600,7 @@ public class DistributionManager implements DM {
*
* @param owner the agent that owns the monitor to remove
*/
+ @Override
public void removeHealthMonitor(InternalDistributedMember owner, int theId) {
final HealthMonitor hm = getHealthMonitor(owner);
if (hm != null && hm.getId() == theId) {
@@ -3992,7 +3609,7 @@ public class DistributionManager implements DM {
}
}
- public void removeAllHealthMonitors() {
+ private void removeAllHealthMonitors() {
Iterator it = this.hmMap.values().iterator();
while (it.hasNext()) {
HealthMonitor hm = (HealthMonitor) it.next();
@@ -4001,12 +3618,13 @@ public class DistributionManager implements DM {
}
}
- // For feature request #32887
- public Set getAdminMemberSet() {
+ @Override
+ public Set<InternalDistributedMember> getAdminMemberSet() {
return this.adminConsoles;
}
/** Returns count of members filling the specified role */
+ @Override
public int getRoleCount(Role role) {
int count = 0;
Set mbrs = getDistributionManagerIds();
@@ -4024,6 +3642,7 @@ public class DistributionManager implements DM {
}
/** Returns true if at least one member is filling the specified role */
+ @Override
public boolean isRolePresent(Role role) {
Set mbrs = getDistributionManagerIds();
for (Iterator mbrIter = mbrs.iterator(); mbrIter.hasNext();) {
@@ -4039,6 +3658,7 @@ public class DistributionManager implements DM {
}
/** Returns a set of all roles currently in the distributed system. */
+ @Override
public Set getAllRoles() {
Set allRoles = new HashSet();
Set mbrs = getDistributionManagerIds();
@@ -4060,6 +3680,7 @@ public class DistributionManager implements DM {
* This method was added to allow hydra to obtain thread-local data for transport from one thread
* to another.
*/
+ @Override
public MembershipManager getMembershipManager() {
// NOTE: do not add cancellation checks here. This method is
// used during auto-reconnect after the DS has been closed
@@ -4265,7 +3886,7 @@ public class DistributionManager implements DM {
* Does cleanup relating to this member. And marks the serial executor associated with this
* member for re-use.
*/
- public void handleMemberDeparture(InternalDistributedMember member) {
+ private void handleMemberDeparture(InternalDistributedMember member) {
Integer queueId = getQueueId(member, false);
if (queueId == null) {
return;
@@ -4298,7 +3919,7 @@ public class DistributionManager implements DM {
}
}
- public void awaitTermination(long time, TimeUnit unit) throws InterruptedException {
+ private void awaitTermination(long time, TimeUnit unit) throws InterruptedException {
long timeNanos = unit.toNanos(time);
long remainingNanos = timeNanos;
long start = System.nanoTime();
@@ -4312,7 +3933,7 @@ public class DistributionManager implements DM {
}
}
- protected void shutdown() {
+ private void shutdown() {
for (Iterator iter = serialQueuedExecutorMap.values().iterator(); iter.hasNext();) {
ExecutorService executor = (ExecutorService) iter.next();
executor.shutdown();
@@ -4342,27 +3963,31 @@ public class DistributionManager implements DM {
* This is the listener implementation for responding from events from the Membership Manager.
*
*/
- private class MyListener implements DistributedMembershipListener {
+ private class DMListener implements DistributedMembershipListener {
DistributionManager dm;
- public MyListener(DistributionManager dm) {
+ public DMListener(DistributionManager dm) {
this.dm = dm;
}
+ @Override
public boolean isShutdownMsgSent() {
return shutdownMsgSent;
}
+ @Override
public void membershipFailure(String reason, Throwable t) {
exceptionInThreads = true;
DistributionManager.this.rootCause = t;
getSystem().disconnect(reason, t, true);
}
+ @Override
public void messageReceived(DistributionMessage message) {
handleIncomingDMsg(message);
}
+ @Override
public void newMemberConnected(InternalDistributedMember member) {
// Do not elect the elder here as surprise members invoke this callback
// without holding the view lock. That can cause a race condition and
@@ -4371,6 +3996,7 @@ public class DistributionManager implements DM {
dm.addNewMember(member);
}
+ @Override
public void memberDeparted(InternalDistributedMember theId, boolean crashed, String reason) {
boolean wasAdmin = getAdminMemberSet().contains(theId);
if (wasAdmin) {
@@ -4384,29 +4010,32 @@ public class DistributionManager implements DM {
message.setAlertListenerExpected(true);
message.setIgnoreAlertListenerRemovalFailure(true); // we don't know if it was a listener so
// don't issue a warning
- message.setRecipient(myid);
+ message.setRecipient(localAddress);
message.setReason(reason); // added for #37950
handleIncomingDMsg(message);
}
dm.handleManagerDeparture(theId, crashed, reason);
}
+ @Override
public void memberSuspect(InternalDistributedMember suspect,
InternalDistributedMember whoSuspected, String reason) {
dm.handleManagerSuspect(suspect, whoSuspected, reason);
}
+ @Override
public void viewInstalled(NetView view) {
processElderSelection();
dm.handleViewInstalled(view);
}
- /** this is invoked when quorum is being lost, before the view has been installed */
+ @Override
public void quorumLost(Set<InternalDistributedMember> failures,
List<InternalDistributedMember> remaining) {
dm.handleQuorumLost(failures, remaining);
}
+ @Override
public DistributionManager getDM() {
return dm;
}
@@ -4435,14 +4064,14 @@ public class DistributionManager implements DM {
return this.id;
}
- public void handleEvent(DistributionManager manager) {
+ void handleEvent(DistributionManager manager) {
handleEvent(manager, manager.membershipListeners.keySet());
handleEvent(manager, manager.allMembershipListeners);
}
protected abstract void handleEvent(MembershipListener listener);
- protected void handleEvent(DistributionManager manager,
+ private void handleEvent(DistributionManager manager,
Set<MembershipListener> membershipListeners) {
for (MembershipListener listener : membershipListeners) {
try {
@@ -4644,6 +4273,7 @@ public class DistributionManager implements DM {
*
* @see org.apache.geode.distributed.internal.DM#getRootCause()
*/
+ @Override
public Throwable getRootCause() {
return this.rootCause;
}
@@ -4653,6 +4283,7 @@ public class DistributionManager implements DM {
*
* @see org.apache.geode.distributed.internal.DM#setRootCause(java.lang.Throwable)
*/
+ @Override
public void setRootCause(Throwable t) {
this.rootCause = t;
}
@@ -4664,10 +4295,12 @@ public class DistributionManager implements DM {
*
* @since GemFire 5.9
*/
+ @Override
public Set<InternalDistributedMember> getMembersInThisZone() {
return getMembersInSameZone(getDistributionManagerId());
}
+ @Override
public Set<InternalDistributedMember> getMembersInSameZone(
InternalDistributedMember targetMember) {
Set<InternalDistributedMember> buddyMembers = new HashSet<InternalDistributedMember>();
@@ -4693,6 +4326,7 @@ public class DistributionManager implements DM {
return buddyMembers;
}
+ @Override
public boolean areInSameZone(InternalDistributedMember member1,
InternalDistributedMember member2) {
@@ -4705,11 +4339,13 @@ public class DistributionManager implements DM {
}
}
+ @Override
public void acquireGIIPermitUninterruptibly() {
this.parallelGIIs.acquireUninterruptibly();
this.stats.incInitialImageRequestsInProgress(1);
}
+ @Override
public void releaseGIIPermit() {
this.stats.incInitialImageRequestsInProgress(-1);
this.parallelGIIs.release();
@@ -4721,36 +4357,30 @@ public class DistributionManager implements DM {
}
}
+ @Override
public int getDistributedSystemId() {
return this.distributedSystemId;
}
/**
- * this causes all members in the system to log thread dumps If useNative is true we attempt to
- * use OSProcess native code for the dumps. This goes to stdout instead of the system.log files.
- */
- public void printDistributedSystemStacks(boolean useNative) {
- printStacks(new HashSet(getDistributionManagerIds()), useNative);
- }
-
- /**
* this causes the given InternalDistributedMembers to log thread dumps. If useNative is true we
* attempt to use OSProcess native code for the dumps. This goes to stdout instead of the
* system.log files.
*/
public void printStacks(Collection ids, boolean useNative) {
Set requiresMessage = new HashSet();
- if (ids.contains(myid)) {
+ if (ids.contains(localAddress)) {
OSProcess.printStacks(0, useNative);
}
if (useNative) {
requiresMessage.addAll(ids);
- ids.remove(myid);
+ ids.remove(localAddress);
} else {
for (Iterator it = ids.iterator(); it.hasNext();) {
InternalDistributedMember mbr = (InternalDistributedMember) it.next();
- if (mbr.getProcessId() > 0 && mbr.getInetAddress().equals(this.myid.getInetAddress())) {
- if (!mbr.equals(myid)) {
+ if (mbr.getProcessId() > 0
+ && mbr.getInetAddress().equals(this.localAddress.getInetAddress())) {
+ if (!mbr.equals(localAddress)) {
if (!OSProcess.printStacks(mbr.getProcessId(), false)) {
requiresMessage.add(mbr);
}
@@ -4766,6 +4396,7 @@ public class DistributionManager implements DM {
}
}
+ @Override
public Set<DistributedMember> getGroupMembers(String group) {
HashSet<DistributedMember> result = null;
for (DistributedMember m : (Set<DistributedMember>) getDistributionManagerIdsIncludingAdmin()) {
@@ -4798,6 +4429,7 @@ public class DistributionManager implements DM {
}
}
+ /** test method to get the member IDs of all locators in the distributed system */
public Set<InternalDistributedMember> getLocatorDistributionManagerIds() {
// access to members synchronized under membersLock in order to
// ensure serialization
@@ -4836,4 +4468,76 @@ public class DistributionManager implements DM {
}
return result;
}
+
+
+ private static class Stopper extends CancelCriterion {
+ private DistributionManager dm;
+
+ Stopper(DistributionManager dm) {
+ this.dm = dm;
+ }
+
+ @Override
+ public String cancelInProgress() {
+ checkFailure();
+
+ // remove call to validateDM() to fix bug 38356
+
+ if (dm.shutdownMsgSent) {
+ return LocalizedStrings.DistributionManager__0_MESSAGE_DISTRIBUTION_HAS_TERMINATED
+ .toLocalizedString(dm.toString());
+ }
+ if (dm.rootCause != null) {
+ return dm.toString() + ": " + dm.rootCause.getMessage();
+ }
+
+ // Nope.
+ return null;
+ }
+
+ @Override
+ public RuntimeException generateCancelledException(Throwable e) {
+ String reason = cancelInProgress();
+ if (reason == null) {
+ return null;
+ }
+ Throwable rc = dm.rootCause; // volatile read
+ if (rc == null) {
+ // No root cause, specify the one given and be done with it.
+ return new DistributedSystemDisconnectedException(reason, e);
+ }
+
+ if (e == null) {
+ // Caller did not specify any root cause, so just use our own.
+ return new DistributedSystemDisconnectedException(reason, rc);
+ }
+
+ // Attempt to stick rootCause at tail end of the exception chain.
+ Throwable nt = e;
+ while (nt.getCause() != null) {
+ nt = nt.getCause();
+ }
+ if (nt == rc) {
+ // Root cause already in place; we're done
+ return new DistributedSystemDisconnectedException(reason, e);
+ }
+
+ try {
+ nt.initCause(rc);
+ return new DistributedSystemDisconnectedException(reason, e);
+ } catch (IllegalStateException e2) {
+ // Bug 39496 (Jrockit related) Give up. The following
+ // error is not entirely sane but gives the correct general picture.
+ return new DistributedSystemDisconnectedException(reason, rc);
+ }
+ }
+ }
+
+ private final Stopper stopper = new Stopper(this);
+
+ @Override
+ public CancelCriterion getCancelCriterion() {
+ return stopper;
+ }
+
}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionMessage.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionMessage.java
index 20884b2..134488b 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionMessage.java
@@ -59,6 +59,14 @@ import org.apache.geode.internal.util.Breadcrumbs;
*/
public abstract class DistributionMessage implements DataSerializableFixedID, Cloneable {
+ /**
+ * WARNING: setting this to true may break dunit tests.
+ * <p>
+ * see org.apache.geode.cache30.ClearMultiVmCallBkDUnitTest
+ */
+ private static final boolean INLINE_PROCESS =
+ !Boolean.getBoolean("DistributionManager.enqueueOrderedMessages");
+
private static final Logger logger = LogService.getLogger();
/**
@@ -410,7 +418,7 @@ public abstract class DistributionMessage implements DataSerializableFixedID, Cl
* Schedule this message's process() method in a thread determined by getExecutor()
*/
protected void schedule(final DistributionManager dm) {
- boolean inlineProcess = DistributionManager.INLINE_PROCESS
+ boolean inlineProcess = INLINE_PROCESS
&& getProcessorType() == DistributionManager.SERIAL_EXECUTOR && !isPreciousThread();
boolean forceInline = this.acker != null || getInlineProcess() || Connection.isDominoThread();
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalDistributedSystem.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalDistributedSystem.java
index 35052d4..4e5c516 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalDistributedSystem.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalDistributedSystem.java
@@ -82,7 +82,6 @@ import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.admin.remote.DistributionLocatorId;
import org.apache.geode.internal.cache.CacheConfig;
import org.apache.geode.internal.cache.CacheServerImpl;
-import org.apache.geode.internal.cache.CacheService;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
@@ -759,7 +758,7 @@ public class InternalDistributedSystem extends DistributedSystem
Assert.assertTrue(this.dm.getSystem() == this);
try {
- this.id = this.dm.getChannelId();
+ this.id = this.dm.getMembershipPort();
} catch (DistributedSystemDisconnectedException e) {
// bug #48144 - The dm's channel threw an NPE. It now throws this exception
// but during startup we should instead throw a SystemConnectException
@@ -1625,7 +1624,6 @@ public class InternalDistributedSystem extends DistributedSystem
* Returns the id of this connection to the distributed system. This is actually the port of the
* distribution manager's distribution channel.
*
- * @see org.apache.geode.distributed.internal.DistributionChannel#getId
*/
@Override
public long getId() {
@@ -3019,7 +3017,6 @@ public class InternalDistributedSystem extends DistributedSystem
public static void setCommandLineAdmin(boolean adminOnly) {
DistributedSystem.setEnableAdministrationOnly(adminOnly);
- DistributionManager.isCommandLineAdminVM = adminOnly;
}
public boolean isServerLocator() {
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/LonerDistributionManager.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/LonerDistributionManager.java
index 142f72c..9dc1ceb 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/LonerDistributionManager.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/LonerDistributionManager.java
@@ -213,7 +213,7 @@ public class LonerDistributionManager implements DM {
return this.elderState;
}
- public long getChannelId() {
+ public long getMembershipPort() {
return 0;
}
@@ -279,6 +279,11 @@ public class LonerDistributionManager implements DM {
return executor;
}
+ @Override
+ public Executor getFunctionExcecutor() {
+ return executor;
+ }
+
public Map getChannelMap() {
return null;
}
@@ -1375,12 +1380,6 @@ public class LonerDistributionManager implements DM {
}
@Override
- public boolean isSharedConfigurationServiceEnabledForDS() {
- // return false for loner
- return false;
- }
-
- @Override
public void setCache(InternalCache instance) {
this.cache = instance;
}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/StartupResponseMessage.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/StartupResponseMessage.java
index 3ba5d31..45ed83c 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/StartupResponseMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/StartupResponseMessage.java
@@ -17,7 +17,6 @@ package org.apache.geode.distributed.internal;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
-import java.util.Properties;
import java.util.Set;
import org.apache.logging.log4j.Logger;
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/MembershipManager.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/MembershipManager.java
index 1a787f3..afcd2a2 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/MembershipManager.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/MembershipManager.java
@@ -88,6 +88,13 @@ public interface MembershipManager {
public boolean isBeingSick();
/**
+ * Instructs this manager to shut down
+ *
+ * @param beforeJoined whether we've joined the cluster or not
+ */
+ public void disconnect(boolean beforeJoined);
+
+ /**
* Instruct this manager to release resources
*/
public void shutdown();
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/mgr/GMSMembershipManager.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/mgr/GMSMembershipManager.java
index 9232c6e..a2fc407 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/mgr/GMSMembershipManager.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/membership/gms/mgr/GMSMembershipManager.java
@@ -2637,4 +2637,13 @@ public class GMSMembershipManager implements MembershipManager, Manager {
return shutdownInProgress || (dm != null && dm.isShutdownStarted());
}
+ public void disconnect(boolean beforeJoined) {
+ if (beforeJoined) {
+ uncleanShutdown("Failed to start distribution", null);
+ } else {
+ shutdown();
+ }
+ }
+
+
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteGfManagerAgent.java b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteGfManagerAgent.java
index 2a18bd8..dfe241d 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteGfManagerAgent.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/RemoteGfManagerAgent.java
@@ -375,7 +375,7 @@ class RemoteGfManagerAgent implements GfManagerAgent {
// ignore a forced disconnect and finish clean-up
}
- if (system != null && DistributionManager.isDedicatedAdminVM && system.isConnected()) {
+ if (system != null && DistributionManager.isDedicatedAdminVM() && system.isConnected()) {
system.disconnect();
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
index 76113f1..d136d20 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
@@ -2689,7 +2689,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
@Override
public Set<DistributedMember> getAdminMembers() {
- return (Set<DistributedMember>) this.dm.getAdminMemberSet();
+ return (Set) this.dm.getAdminMemberSet();
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
index a0b003c..abb0d09 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
@@ -2807,7 +2807,7 @@ public class PartitionedRegion extends LocalRegion
// given that most manipulation of values is remote (requiring serialization to send).
// But... function execution always implies local manipulation of
// values so keeping locally updated values in Object form should be more efficient.
- if (!DistributionManager.isFunctionExecutionThread.get()) {
+ if (!DistributionManager.isFunctionExecutionThread()) {
// TODO: this condition may not help since BucketRegion.virtualPut calls
// forceSerialized
br.forceSerialized(event);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/SearchLoadAndWriteProcessor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/SearchLoadAndWriteProcessor.java
index 5195362..83fa818 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/SearchLoadAndWriteProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/SearchLoadAndWriteProcessor.java
@@ -2149,8 +2149,8 @@ public class SearchLoadAndWriteProcessor implements MembershipListener {
msg.setRecipient(recipient);
try {
- processor.distributionManager.putOutgoingUserData(msg);
- } catch (NotSerializableException ignore) {
+ processor.distributionManager.putOutgoing(msg);
+ } catch (InternalGemFireException e) {
throw new IllegalArgumentException(
LocalizedStrings.SearchLoadAndWriteProcessor_MESSAGE_NOT_SERIALIZABLE
.toLocalizedString());
diff --git a/geode-core/src/test/java/org/apache/geode/admin/AdminTestHelper.java b/geode-core/src/test/java/org/apache/geode/admin/AdminTestHelper.java
index e8b51f8..b1531f7 100644
--- a/geode-core/src/test/java/org/apache/geode/admin/AdminTestHelper.java
+++ b/geode-core/src/test/java/org/apache/geode/admin/AdminTestHelper.java
@@ -22,22 +22,22 @@ public class AdminTestHelper {
private AdminTestHelper() {}
public static void checkEnableAdministrationOnly(boolean v, boolean expectException) {
- boolean origIsDedicatedAdminVM = DistributionManager.isDedicatedAdminVM;
+ boolean origIsDedicatedAdminVM = DistributionManager.isDedicatedAdminVM();
if (expectException) {
try {
AdminDistributedSystemFactory.setEnableAdministrationOnly(v);
fail("expected IllegalStateException");
} catch (IllegalStateException expected) {
- assertEquals(origIsDedicatedAdminVM, DistributionManager.isDedicatedAdminVM);
+ assertEquals(origIsDedicatedAdminVM, DistributionManager.isDedicatedAdminVM());
} finally {
- DistributionManager.isDedicatedAdminVM = origIsDedicatedAdminVM;
+ DistributionManager.setIsDedicatedAdminVM(origIsDedicatedAdminVM);
}
} else {
try {
AdminDistributedSystemFactory.setEnableAdministrationOnly(v);
- assertEquals(v, DistributionManager.isDedicatedAdminVM);
+ assertEquals(v, DistributionManager.isDedicatedAdminVM());
} finally {
- DistributionManager.isDedicatedAdminVM = origIsDedicatedAdminVM;
+ DistributionManager.setIsDedicatedAdminVM(origIsDedicatedAdminVM);
}
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/cache/ConnectionPoolDUnitTest.java b/geode-core/src/test/java/org/apache/geode/cache/ConnectionPoolDUnitTest.java
index f39a147..e7110fb 100755
--- a/geode-core/src/test/java/org/apache/geode/cache/ConnectionPoolDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/cache/ConnectionPoolDUnitTest.java
@@ -20,7 +20,6 @@ import static org.junit.runners.MethodSorters.*;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
@@ -62,7 +61,6 @@ import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifierStats;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.logging.LocalLogWriter;
-import org.apache.geode.internal.net.SocketCreatorFactory;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.Invoke;
@@ -908,9 +906,9 @@ public class ConnectionPoolDUnitTest extends JUnit4CacheTestCase {
List l = InternalDistributedSystem.getExistingSystems();
if (l.size() > 1) {
getSystem().getLogWriter().info("validateDS: size=" + l.size() + " isDedicatedAdminVM="
- + DistributionManager.isDedicatedAdminVM + " l=" + l);
+ + DistributionManager.isDedicatedAdminVM() + " l=" + l);
}
- assertFalse(DistributionManager.isDedicatedAdminVM);
+ assertFalse(DistributionManager.isDedicatedAdminVM());
assertEquals(1, l.size());
}
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/SystemAdminDUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/SystemAdminDUnitTest.java
index 90ddefa..8639cf5 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/SystemAdminDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/SystemAdminDUnitTest.java
@@ -86,8 +86,7 @@ public class SystemAdminDUnitTest extends JUnit4DistributedTestCase {
} finally {
// SystemAdmin calls methods that set these static variables
- DistributionManager.isDedicatedAdminVM = false;
- DistributionManager.isCommandLineAdminVM = false;
+ DistributionManager.setIsDedicatedAdminVM(false);
SystemAdmin.setDistributedSystemProperties(null);
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/ConsoleDistributionManagerDUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/ConsoleDistributionManagerDUnitTest.java
index e2ed714..9b88435 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/ConsoleDistributionManagerDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/ConsoleDistributionManagerDUnitTest.java
@@ -85,7 +85,7 @@ public class ConsoleDistributionManagerDUnitTest extends JUnit4CacheTestCase
firstTime = false;
}
- DistributionManager.isDedicatedAdminVM = true;
+ DistributionManager.setIsDedicatedAdminVM(true);
populateCache();
@@ -138,7 +138,7 @@ public class ConsoleDistributionManagerDUnitTest extends JUnit4CacheTestCase
throw e;
} catch (Throwable ignore) {
}
- DistributionManager.isDedicatedAdminVM = false;
+ DistributionManager.setIsDedicatedAdminVM(false);
}
}
}
@@ -153,7 +153,7 @@ public class ConsoleDistributionManagerDUnitTest extends JUnit4CacheTestCase
try {
disconnectFromDS(); // make sure there's no ldm lying around
} finally {
- DistributionManager.isDedicatedAdminVM = false;
+ DistributionManager.setIsDedicatedAdminVM(false);
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/InternalDistributedSystemJUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/InternalDistributedSystemJUnitTest.java
index dd4147a..9cac77c 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/InternalDistributedSystemJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/InternalDistributedSystemJUnitTest.java
@@ -32,7 +32,6 @@ import java.util.logging.Level;
import org.junit.After;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -67,7 +66,7 @@ public class InternalDistributedSystemJUnitTest {
* Creates a <code>DistributedSystem</code> with the given configuration properties.
*/
protected InternalDistributedSystem createSystem(Properties props) {
- assertFalse(org.apache.geode.distributed.internal.DistributionManager.isDedicatedAdminVM);
+ assertFalse(DistributionManager.isDedicatedAdminVM());
this.system = (InternalDistributedSystem) DistributedSystem.connect(props);
return this.system;
}
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/MembershipJUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/MembershipJUnitTest.java
index 477e0c5..2b22b64 100755
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/MembershipJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/MembershipJUnitTest.java
@@ -222,7 +222,7 @@ public class MembershipJUnitTest {
// let the managers idle for a while and get used to each other
// Thread.sleep(4000l);
- m2.shutdown();
+ m2.disconnect(false);
assertTrue(!m2.isConnected());
assertTrue(m1.getView().size() == 1);
@@ -357,7 +357,7 @@ public class MembershipJUnitTest {
// let the managers idle for a while and get used to each other
Thread.sleep(4000l);
- m2.shutdown();
+ m2.disconnect(false);
assertTrue(!m2.isConnected());
assertTrue(m1.getView().size() == 1);
@@ -365,10 +365,10 @@ public class MembershipJUnitTest {
} finally {
if (m2 != null) {
- m2.shutdown();
+ m2.disconnect(false);
}
if (m1 != null) {
- m1.shutdown();
+ m1.disconnect(false);
}
if (l != null) {
l.stop();
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java
index 46fffa7..65de196 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/gms/locator/GMSLocatorRecoveryJUnitTest.java
@@ -185,7 +185,7 @@ public class GMSLocatorRecoveryJUnitTest {
l2.getMembers().contains(m1.getLocalMember()));
} finally {
if (m1 != null) {
- m1.shutdown();
+ m1.disconnect(false);
}
if (l != null) {
l.stop();
--
To stop receiving notification emails like this one, please contact
['"commits@geode.apache.org" <co...@geode.apache.org>'].