You are viewing a plain text version of this content. The canonical link for it is here.
Posted to java-dev@axis.apache.org by az...@apache.org on 2007/06/18 17:05:34 UTC
svn commit: r548375 - in /webservices/axis2/trunk/java/modules:
clustering/src/org/apache/axis2/clustering/context/
clustering/src/org/apache/axis2/clustering/tribes/
kernel/src/org/apache/axis2/clustering/context/
Author: azeez
Date: Mon Jun 18 08:05:30 2007
New Revision: 548375
URL: http://svn.apache.org/viewvc?view=rev&rev=548375
Log:
Keeping track of the mebers to whom messages have been retransmitted
Modified:
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/context/ContextClusteringCommandFactory.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/AckManager.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ChannelListener.java
webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/context/Replicator.java
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/context/ContextClusteringCommandFactory.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/context/ContextClusteringCommandFactory.java?view=diff&rev=548375&r1=548374&r2=548375
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/context/ContextClusteringCommandFactory.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/context/ContextClusteringCommandFactory.java Mon Jun 18 08:05:30 2007
@@ -16,9 +16,17 @@
package org.apache.axis2.clustering.context;
import org.apache.axiom.om.util.UUIDGenerator;
-import org.apache.axis2.clustering.context.commands.*;
+import org.apache.axis2.clustering.context.commands.ContextClusteringCommandCollection;
+import org.apache.axis2.clustering.context.commands.UpdateConfigurationContextCommand;
+import org.apache.axis2.clustering.context.commands.UpdateContextCommand;
+import org.apache.axis2.clustering.context.commands.UpdateServiceContextCommand;
+import org.apache.axis2.clustering.context.commands.UpdateServiceGroupContextCommand;
import org.apache.axis2.clustering.tribes.AckManager;
-import org.apache.axis2.context.*;
+import org.apache.axis2.context.AbstractContext;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.context.PropertyDifference;
+import org.apache.axis2.context.ServiceContext;
+import org.apache.axis2.context.ServiceGroupContext;
import org.apache.axis2.deployment.DeploymentConstants;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -98,6 +106,8 @@
AckManager.addInitialAcknowledgement(cmd);
}
}
+
+ //TODO: This is causing a concurrency issue. Need to look into this
context.clearPropertyDifferences(); // Once we send the diffs, we should clear the diffs
return cmd;
}
@@ -117,31 +127,31 @@
// Sometimes, there can be failures, so if an exception occurs, we retry
// TODO: Need to investigate these failures. Looks like this is becuase the contexts are not synchronized
- while (true) {
- Map diffs = context.getPropertyDifferences();
- try {
- for (Iterator iter = diffs.keySet().iterator(); iter.hasNext();) {
- String key = (String) iter.next();
- Object prop = context.getPropertyNonReplicable(key);
-
- // First check whether it is serializable
- if (prop instanceof Serializable) {
-
- // Next check whether it matches an excluded pattern
- if (!isExcluded(key,
- context.getClass().getName(),
- excludedPropertyPatterns)) {
- log.debug("sending property =" + key + "-" + prop);
- PropertyDifference diff = (PropertyDifference) diffs.get(key);
- diff.setValue(prop);
- updateCmd.addProperty(diff);
- }
- }
+// while (true) {
+ Map diffs = context.getPropertyDifferences();
+// try {
+ for (Iterator iter = diffs.keySet().iterator(); iter.hasNext();) {
+ String key = (String) iter.next();
+ Object prop = context.getPropertyNonReplicable(key);
+
+ // First check whether it is serializable
+ if (prop instanceof Serializable) {
+
+ // Next check whether it matches an excluded pattern
+ if (!isExcluded(key,
+ context.getClass().getName(),
+ excludedPropertyPatterns)) {
+ log.debug("sending property =" + key + "-" + prop);
+ PropertyDifference diff = (PropertyDifference) diffs.get(key);
+ diff.setValue(prop);
+ updateCmd.addProperty(diff);
}
- break;
- } catch (Exception ignored) {
}
}
+// break;
+// } catch (Exception ignored) {
+// }
+// }
} else {
for (Iterator iter = context.getPropertyNames(); iter.hasNext();) {
String key = (String) iter.next();
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/AckManager.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/AckManager.java?view=diff&rev=548375&r1=548374&r2=548375
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/AckManager.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/AckManager.java Mon Jun 18 08:05:30 2007
@@ -21,6 +21,7 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
@@ -41,9 +42,8 @@
String memberId) {
MessageACK ack = (MessageACK) messageAckTable.get(messageUniqueId);
if (ack != null) {
- List memberList = ack.getMemberList();
- if (!memberList.contains(memberId)) { // If the member has not already ACKed
- memberList.add(memberId);
+ if (!ack.hasACKed(memberId)) { // If the member has not already ACKed
+ ack.addACK(memberId);
}
}
}
@@ -53,7 +53,6 @@
boolean isAcknowledged = false;
MessageACK ack = (MessageACK) messageAckTable.get(messageUniqueId);
- List memberList = ack.getMemberList();
// Check that all members in the memberList are same as the total member list,
// which will indicate that all members have ACKed the message
@@ -64,14 +63,15 @@
for (int i = 0; i < members.length; i++) {
Member member = members[i];
String memberHost = TribesUtil.getHost(member);
- if (member.isReady() && !memberList.contains(memberHost)) {
+ if (member.isReady() && !ack.hasACKed(memberHost)) {
log.debug("[NO ACK] from member " + memberHost);
- log.debug("ACKed member list=" + memberList);
// If a new member joined the cluster recently,
// we need to retransmit the message to this member, if an ACK has not been
- // received from this member.
- if (member.getMemberAliveTime() < 1000) { // TODO: Check
+ // received from this member. We retransmit only once.
+ if (member.getMemberAliveTime() < 5000 &&
+ !ack.isRestransmittedToMember(memberHost)) { // TODO: Check
+
sender.sendToMember(ack.getCommand(), member);
log.debug("Retransimitting msg " + ack.getCommand().getUniqueId() +
" to member " + memberHost);
@@ -95,17 +95,30 @@
private static class MessageACK {
private ContextClusteringCommand command;
private List memberList = new Vector();
+ private List retransmittedList = new ArrayList();
public MessageACK(ContextClusteringCommand command) {
this.command = command;
}
+ public void addACK(String memberId) {
+ memberList.add(memberId);
+ }
+
public ContextClusteringCommand getCommand() {
return command;
}
- public List getMemberList() {
- return memberList;
+ public boolean hasACKed(String memberId) {
+ return memberList.contains(memberId);
+ }
+
+ public void addToRestransmittedList(String memberId) {
+ retransmittedList.add(memberId);
+ }
+
+ public boolean isRestransmittedToMember(String memberId) {
+ return retransmittedList.contains(memberId);
}
}
}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ChannelListener.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ChannelListener.java?view=diff&rev=548375&r1=548374&r2=548375
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ChannelListener.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ChannelListener.java Mon Jun 18 08:05:30 2007
@@ -35,7 +35,6 @@
import java.io.Serializable;
-
public class ChannelListener implements org.apache.catalina.tribes.ChannelListener {
private static final Log log = LogFactory.getLog(ChannelListener.class);
@@ -132,7 +131,6 @@
private void processMessage(Serializable msg, Member sender) throws ClusteringFault {
- long start = System.currentTimeMillis();
if (msg instanceof ContextClusteringCommand && contextManager != null) {
ContextClusteringCommand ctxCmd = (ContextClusteringCommand) msg;
contextManager.process(ctxCmd);
@@ -153,7 +151,6 @@
controlCommandProcessor.process((ControlCommand) msg,
sender);
}
- System.err.println("######### Time to process=" + (System.currentTimeMillis() - start));
}
/**
Modified: webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/context/Replicator.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/context/Replicator.java?view=diff&rev=548375&r1=548374&r2=548375
==============================================================================
--- webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/context/Replicator.java (original)
+++ webservices/axis2/trunk/java/modules/kernel/src/org/apache/axis2/clustering/context/Replicator.java Mon Jun 18 08:05:30 2007
@@ -133,9 +133,9 @@
// Do the actual replication here
if (!contexts.isEmpty()) {
- String msgUUID =
- contextManager.updateContexts((AbstractContext[]) contexts.
- toArray(new AbstractContext[contexts.size()]));
+ AbstractContext[] contextArray =
+ (AbstractContext[]) contexts.toArray(new AbstractContext[contexts.size()]);
+ String msgUUID = contextManager.updateContexts(contextArray);
waitForACKs(contextManager, msgUUID, msgContext.getRootContext());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: axis-cvs-unsubscribe@ws.apache.org
For additional commands, e-mail: axis-cvs-help@ws.apache.org