6:06 p.m.
Author: manik.surtani(a)jboss.com
Date: 2007-11-29 12:06:22 -0500 (Thu, 29 Nov 2007)
New Revision: 4790
Modified:
core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java
core/trunk/src/main/java/org/jboss/cache/loader/ClusteredCacheLoader.java
Log:
JBCACHE-1227 - group requests hanging when no node is able to provide desired response
Modified:
core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java
===================================================================
---
core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java 2007-11-29
16:58:27 UTC (rev 4789)
+++
core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java 2007-11-29
17:06:22 UTC (rev 4790)
@@ -321,7 +321,7 @@
MethodCall dGrav =
MethodCallFactory.create(MethodDeclarations.dataGravitationMethod, fqn, searchSubtrees);
// doing a GET_ALL is crappy but necessary since JGroups' GET_FIRST could
return null results from nodes that do
// not have either the primary OR backup, and stop polling other valid nodes.
- List resps = cache.getRPCManager().callRemoteMethods(mbrs, dGrav,
GroupRequest.GET_FIRST, true, buddyManager.getBuddyCommunicationTimeout(), new
ResponseValidityFilter());
+ List resps = cache.getRPCManager().callRemoteMethods(mbrs, dGrav,
GroupRequest.GET_ALL, true, buddyManager.getBuddyCommunicationTimeout(), new
ResponseValidityFilter(mbrs, cache.getLocalAddress()));
if (log.isTraceEnabled())
{
log.trace("got responses " + resps);
@@ -486,20 +486,32 @@
public static class ResponseValidityFilter implements RspFilter
{
private int numValidResponses = 0;
+ private List<Address> pendingResponders;
+ public ResponseValidityFilter(List<Address> expected, Address localAddress)
+ {
+ // so for now I used a list to keep it consistent
+ this.pendingResponders = new ArrayList<Address>(expected);
+ // We'll never get a response from ourself
+ this.pendingResponders.remove(localAddress);
+ }
+
public boolean isAcceptable(Object object, Address address)
{
- if (!(object instanceof GravitateResult)) return false;
+ pendingResponders.remove(address);
- GravitateResult response = (GravitateResult) object;
- if (response.isDataFound()) numValidResponses++;
-
- return response.isDataFound();
+ if (object instanceof GravitateResult)
+ {
+ GravitateResult response = (GravitateResult) object;
+ if (response.isDataFound()) numValidResponses++;
+ }
+ // always return true to make sure a response is logged by the JGroups
RpcDispatcher.
+ return true;
}
public boolean needMoreResponses()
{
- return numValidResponses < 1;
+ return numValidResponses < 1 && pendingResponders.size() > 0;
}
}
}
Modified: core/trunk/src/main/java/org/jboss/cache/loader/ClusteredCacheLoader.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/loader/ClusteredCacheLoader.java 2007-11-29
16:58:27 UTC (rev 4789)
+++ core/trunk/src/main/java/org/jboss/cache/loader/ClusteredCacheLoader.java 2007-11-29
17:06:22 UTC (rev 4790)
@@ -24,6 +24,7 @@
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -90,7 +91,7 @@
MethodCall clusteredGet =
MethodCallFactory.create(MethodDeclarations.clusteredGetMethod, call, false);
List resps = null;
// JBCACHE-1186
- resps = cache.getRPCManager().callRemoteMethods(mbrs, clusteredGet,
GroupRequest.GET_FIRST, true, config.getTimeout(), new ResponseValidityFilter());
+ resps = cache.getRPCManager().callRemoteMethods(mbrs, clusteredGet,
GroupRequest.GET_ALL, true, config.getTimeout(), new ResponseValidityFilter(mbrs,
cache.getLocalAddress()));
if (resps == null)
{
@@ -327,21 +328,34 @@
public static class ResponseValidityFilter implements RspFilter
{
private int numValidResponses = 0;
+ private List<Address> pendingResponders;
+ public ResponseValidityFilter(List<Address> expected, Address localAddress)
+ {
+ this.pendingResponders = new ArrayList<Address>(expected);
+ // We'll never get a response from ourself
+ this.pendingResponders.remove(localAddress);
+ }
+
public boolean isAcceptable(Object object, Address address)
{
- if (!(object instanceof List)) return false;
+ pendingResponders.remove(address);
- List response = (List) object;
- Boolean foundResult = (Boolean) response.get(0);
- if (foundResult) numValidResponses++;
-
- return foundResult;
+ if (object instanceof List)
+ {
+ List response = (List) object;
+ Boolean foundResult = (Boolean) response.get(0);
+ if (foundResult) numValidResponses++;
+ }
+ // always return true to make sure a response is logged by the JGroups
RpcDispatcher.
+ return true;
}
public boolean needMoreResponses()
{
- return numValidResponses < 1;
+ return numValidResponses < 1 && pendingResponders.size() > 0;
}
+
}
+
}