[jbosscache-commits] JBoss Cache SVN: r4502 - in core/trunk/src: main/java/org/jboss/cache/loader and 1 other directories.
jbosscache-commits at lists.jboss.org
jbosscache-commits at lists.jboss.org
Tue Sep 25 09:05:59 EDT 2007
Author: manik.surtani at jboss.com
Date: 2007-09-25 09:05:59 -0400 (Tue, 25 Sep 2007)
New Revision: 4502
Modified:
core/trunk/src/main/java/org/jboss/cache/CacheImpl.java
core/trunk/src/main/java/org/jboss/cache/RPCManager.java
core/trunk/src/main/java/org/jboss/cache/RPCManagerImpl.java
core/trunk/src/main/java/org/jboss/cache/loader/ClusteredCacheLoader.java
core/trunk/src/test/java/org/jboss/cache/loader/ClusteredCacheLoaderTest.java
Log:
JBCACHE-1186 - Use a JGroups RspFilter to filter responses for the clustered cache loader. Note that this imposes a dependency on JGroups 2.6.
Modified: core/trunk/src/main/java/org/jboss/cache/CacheImpl.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/CacheImpl.java 2007-09-24 20:54:55 UTC (rev 4501)
+++ core/trunk/src/main/java/org/jboss/cache/CacheImpl.java 2007-09-25 13:05:59 UTC (rev 4502)
@@ -61,6 +61,7 @@
import org.jgroups.View;
import org.jgroups.blocks.GroupRequest;
import org.jgroups.blocks.RpcDispatcher;
+import org.jgroups.blocks.RspFilter;
import org.jgroups.util.Rsp;
import org.jgroups.util.RspList;
import org.jgroups.util.Util;
@@ -472,7 +473,7 @@
if (log.isWarnEnabled())
{
log.warn("Cannot fetch partial state, targets are " + sources +
- " and stateId is " + stateId);
+ " and stateId is " + stateId);
}
return;
}
@@ -555,7 +556,7 @@
* @throws Error
*/
private void handleLifecycleTransitionFailure(Throwable t)
- throws CacheException, RuntimeException, Error
+ throws CacheException, RuntimeException, Error
{
cacheStatus = CacheStatus.FAILED;
if (t instanceof CacheException)
@@ -839,7 +840,7 @@
catch (CacheException e)
{
log.warn("Needed to call stop() before destroying but stop() " +
- "threw exception. Proceeding to destroy", e);
+ "threw exception. Proceeding to destroy", e);
}
}
else
@@ -913,7 +914,7 @@
if (failed)
{
log.warn("Attempted to stop() from FAILED state, " +
- "but caught exception; try calling destroy()", t);
+ "but caught exception; try calling destroy()", t);
}
handleLifecycleTransitionFailure(t);
}
@@ -1044,7 +1045,7 @@
protected void createEvictionPolicy()
{
if (configuration.getEvictionConfig() != null
- && configuration.getEvictionConfig().isValidConfig())
+ && configuration.getEvictionConfig().isValidConfig())
{
regionManager.setEvictionConfig(configuration.getEvictionConfig());
regionManager.setUsingEvictions(true);
@@ -1146,8 +1147,8 @@
try
{
child = factory.createDataNode(name,
- subtree.getAncestor(i + 1),
- parent, null, true);
+ subtree.getAncestor(i + 1),
+ parent, null, true);
parent.addChild(name, child);
}
finally
@@ -1241,7 +1242,7 @@
if (broken && log.isTraceEnabled())
{
log.trace("Broke lock for node " + node.getFqn() +
- " held by " + deadOwner);
+ " held by " + deadOwner);
}
}
@@ -1293,7 +1294,7 @@
else
{
throw new CacheException("Initial state transfer failed: " +
- "Channel.getState() returned false");
+ "Channel.getState() returned false");
}
}
}
@@ -1422,7 +1423,7 @@
if (log.isTraceEnabled())
{
log.trace(new StringBuffer("_get(").append("\"").append(fqn).append("\", \"").append(key).append("\", \"").
- append(sendNodeEvent).append("\")"));
+ append(sendNodeEvent).append("\")"));
}
if (sendNodeEvent) notifier.notifyNodeVisited(fqn, true, ctx);
NodeSPI<K, V> n = findNode(fqn);
@@ -1848,8 +1849,9 @@
for (Object c : s)
{
NodeSPI child = (NodeSPI) c;
- if (!child.isDeleted()) {
- E e = (E)child.getFqn().getLastElement();
+ if (!child.isDeleted())
+ {
+ E e = (E) child.getFqn().getLastElement();
childNames.add(e);
}
}
@@ -2023,11 +2025,17 @@
@Deprecated
public List callRemoteMethods(List<Address> mbrs, MethodCall method_call,
boolean synchronous, boolean exclude_self, long timeout)
- throws Exception
+ throws Exception
{
return callRemoteMethods(mbrs, method_call, synchronous ? GroupRequest.GET_ALL : GroupRequest.GET_NONE, exclude_self, timeout);
}
+ @Deprecated
+ public List callRemoteMethods(List<Address> mbrs, MethodCall method_call, int mode, boolean exclude_self, long timeout)
+ throws Exception
+ {
+ return callRemoteMethods(mbrs, method_call, mode, exclude_self, timeout, null);
+ }
/**
* Overloaded to allow a finer grained control over JGroups mode
@@ -2042,8 +2050,8 @@
* @deprecated Note this is due to be moved to an interceptor.
*/
@Deprecated
- public List callRemoteMethods(List<Address> mbrs, MethodCall method_call, int mode, boolean exclude_self, long timeout)
- throws Exception
+ public List callRemoteMethods(List<Address> mbrs, MethodCall method_call, int mode, boolean exclude_self, long timeout, RspFilter rspFilter)
+ throws Exception
{
RspList rsps = null;
Rsp rsp;
@@ -2091,12 +2099,10 @@
if (channel.flushSupported())
{
flushBlockGate.await(configuration.getStateRetrievalTimeout());
- rsps = disp.callRemoteMethods(validMembers, method_call, mode, timeout, buddyManager != null && buddyManager.isEnabled());
}
- else
- {
- rsps = disp.callRemoteMethods(validMembers, method_call, mode, timeout, buddyManager != null && buddyManager.isEnabled());
- }
+ rsps = rspFilter == null
+ ? disp.callRemoteMethods(validMembers, method_call, mode, timeout, buddyManager != null && buddyManager.isEnabled())
+ : disp.callRemoteMethods(validMembers, method_call, mode, timeout, buddyManager != null && buddyManager.isEnabled(), false, rspFilter);
// a null response is 99% likely to be due to a marshalling problem - we throw a NSE, this needs to be changed when
// JGroups supports http://jira.jboss.com/jira/browse/JGRP-193
@@ -2154,7 +2160,7 @@
@Deprecated
public List callRemoteMethods(List<Address> members, Method method, Object[] args,
boolean synchronous, boolean exclude_self, long timeout)
- throws Exception
+ throws Exception
{
return callRemoteMethods(members, MethodCallFactory.create(method, args), synchronous, exclude_self, timeout);
}
@@ -2175,7 +2181,7 @@
public List callRemoteMethods(Vector<Address> members, String method_name,
Class[] types, Object[] args,
boolean synchronous, boolean exclude_self, long timeout)
- throws Exception
+ throws Exception
{
Method method = getClass().getDeclaredMethod(method_name, types);
return callRemoteMethods(members, method, args, synchronous, exclude_self, timeout);
@@ -2241,7 +2247,7 @@
* Otherwise they will not be created (used by rollback()).
*/
public void _put(GlobalTransaction tx, String fqn, Map<K, V> data, boolean create_undo_ops)
- throws CacheException
+ throws CacheException
{
_put(tx, Fqn.fromString(fqn), data, create_undo_ops);
}
@@ -2263,7 +2269,7 @@
* Otherwise they will not be created (used by rollback()).
*/
public void _put(GlobalTransaction tx, Fqn fqn, Map<K, V> data, boolean create_undo_ops)
- throws CacheException
+ throws CacheException
{
_put(tx, fqn, data, create_undo_ops, false);
}
@@ -2285,7 +2291,7 @@
* Otherwise they will not be created (used by rollback()).
*/
public void _put(GlobalTransaction tx, Fqn fqn, Map<K, V> data, boolean create_undo_ops, boolean erase_contents)
- throws CacheException
+ throws CacheException
{
if (log.isTraceEnabled())
{
@@ -2320,7 +2326,7 @@
* @return Previous value (if any)
*/
public Object _put(GlobalTransaction tx, String fqn, K key, V value, boolean create_undo_ops)
- throws CacheException
+ throws CacheException
{
return _put(tx, Fqn.fromString(fqn), key, value, create_undo_ops);
}
@@ -2330,9 +2336,9 @@
try
{
return tx != null && (
- tx.getStatus() == javax.transaction.Status.STATUS_ROLLEDBACK ||
- tx.getStatus() == javax.transaction.Status.STATUS_ROLLING_BACK ||
- tx.getStatus() == javax.transaction.Status.STATUS_MARKED_ROLLBACK);
+ tx.getStatus() == javax.transaction.Status.STATUS_ROLLEDBACK ||
+ tx.getStatus() == javax.transaction.Status.STATUS_ROLLING_BACK ||
+ tx.getStatus() == javax.transaction.Status.STATUS_MARKED_ROLLBACK);
}
catch (Exception e)
{
@@ -2347,12 +2353,12 @@
* @return Previous value (if any)
*/
public Object _put(GlobalTransaction tx, Fqn fqn, K key, V value, boolean create_undo_ops)
- throws CacheException
+ throws CacheException
{
if (log.isTraceEnabled())
{
log.trace(new StringBuffer("_put(").append(tx).append(", \"").
- append(fqn).append("\", k=").append(key).append(", v=").append(value).append(")"));
+ append(fqn).append("\", k=").append(key).append(", v=").append(value).append(")"));
}
@@ -2407,7 +2413,7 @@
}
public boolean _remove(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent)
- throws CacheException
+ throws CacheException
{
return _remove(tx, fqn, create_undo_ops, sendNodeEvent, false);
}
@@ -2421,7 +2427,7 @@
* @param sendNodeEvent
*/
public boolean _remove(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent, boolean eviction)
- throws CacheException
+ throws CacheException
{
return _remove(tx, fqn, create_undo_ops, sendNodeEvent, eviction, null);
}
@@ -2441,7 +2447,7 @@
* @throws CacheException
*/
public boolean _remove(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent, boolean eviction, DataVersion version)
- throws CacheException
+ throws CacheException
{
NodeSPI<K, V> n;
@@ -2551,7 +2557,7 @@
* @return Object
*/
public V _remove(GlobalTransaction tx, String fqn, K key, boolean create_undo_ops)
- throws CacheException
+ throws CacheException
{
return _remove(tx, Fqn.fromString(fqn), key, create_undo_ops);
}
@@ -2564,7 +2570,7 @@
* @return Object
*/
public V _remove(GlobalTransaction tx, Fqn fqn, K key, boolean create_undo_ops)
- throws CacheException
+ throws CacheException
{
MethodCall undo_op = null;
V old_value = null;
@@ -2609,7 +2615,7 @@
* Internal method to remove data from a node.
*/
public void _removeData(GlobalTransaction tx, String fqn, boolean create_undo_ops)
- throws CacheException
+ throws CacheException
{
_removeData(tx, Fqn.fromString(fqn), create_undo_ops);
}
@@ -2618,7 +2624,7 @@
* Internal method to remove data from a node.
*/
public void _removeData(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops)
- throws CacheException
+ throws CacheException
{
_removeData(tx, fqn, create_undo_ops, true);
}
@@ -2627,7 +2633,7 @@
* Internal method to remove data from a node.
*/
public void _removeData(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent)
- throws CacheException
+ throws CacheException
{
_removeData(tx, fqn, create_undo_ops, sendNodeEvent, false);
}
@@ -2636,7 +2642,7 @@
* Internal method to remove data from a node.
*/
public void _removeData(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent, boolean eviction)
- throws CacheException
+ throws CacheException
{
_removeData(tx, fqn, create_undo_ops, sendNodeEvent, eviction, null);
}
@@ -2645,7 +2651,7 @@
* Internal method to remove data from a node.
*/
public void _removeData(GlobalTransaction tx, Fqn fqn, boolean create_undo_ops, boolean sendNodeEvent, boolean eviction, DataVersion version)
- throws CacheException
+ throws CacheException
{
MethodCall undo_op = null;
@@ -2673,7 +2679,7 @@
if (!data.isEmpty())
{
undo_op = MethodCallFactory.create(MethodDeclarations.putDataMethodLocal,
- tx, fqn, new HashMap<K,V>(data), false);
+ tx, fqn, new HashMap<K, V>(data), false);
}
}
@@ -2790,10 +2796,10 @@
/**
* Very much like an evict(), except that regardless of whether there is a child present, this call will never
* remove the node from memory - just remove its contents.
- *
+ * <p/>
* Also, potentially throws a cache exception if data versioning is used and the node in memory has a newer data
* version than what is passed in.
- *
+ * <p/>
* Finally, the data version of the in-memory node is updated to the version being evicted to prevent versions
* going out of sync.
*
@@ -2823,7 +2829,7 @@
put(fqn, m);
getInvocationContext().getOptionOverrides().setCacheModeLocal(false);
NodeSPI nodeSPI = (NodeSPI) root.getChild(fqn);
- nodeSPI.setVersion(versionToInvalidate);
+ nodeSPI.setVersion(versionToInvalidate);
}
}
@@ -2846,7 +2852,7 @@
* Compensating method to {@link #_remove(GlobalTransaction,Fqn,boolean)}.
*/
public void _addChild(GlobalTransaction gtx, Fqn parent_fqn, Object child_name, Node cn, boolean undoOps)
- throws CacheException
+ throws CacheException
{
NodeSPI childNode = (NodeSPI) cn;
if (log.isTraceEnabled())
@@ -2973,7 +2979,7 @@
* @return a GravitateResult which contains the data for the gravitation
*/
public GravitateResult gravitateData(Fqn fqn, boolean searchSubtrees)
- throws CacheException
+ throws CacheException
{
// we need to get the state for this Fqn and its sub-nodes.
@@ -3189,7 +3195,7 @@
* Should not be called.
*/
public void _lock(Fqn fqn, NodeLock.LockType lock_type, boolean recursive)
- throws TimeoutException, LockingException
+ throws TimeoutException, LockingException
{
throw new UnsupportedOperationException("method _lock() should not be invoked on CacheImpl");
}
@@ -3464,7 +3470,7 @@
else
{
log.error("Caught " + t.getClass().getName()
- + " while responding to state transfer request", t);
+ + " while responding to state transfer request", t);
}
}
@@ -3546,7 +3552,7 @@
out = new MarshalledValueOutputStream(baos);
getStateTransferManager().getState(out, Fqn.fromString(sourceRoot),
- configuration.getStateRetrievalTimeout(), true, true);
+ configuration.getStateRetrievalTimeout(), true, true);
}
catch (Throwable t)
{
@@ -4215,18 +4221,18 @@
protected String getDefaultProperties()
{
return "UDP(mcast_addr=224.0.0.36;mcast_port=55566;ip_ttl=32;" +
- "mcast_send_buf_size=150000;mcast_recv_buf_size=80000):" +
- "PING(timeout=1000;num_initial_members=2):" +
- "MERGE2(min_interval=5000;max_interval=10000):" +
- "FD_SOCK:" +
- "VERIFY_SUSPECT(timeout=1500):" +
- "pbcast.NAKACK(gc_lag=50;retransmit_timeout=600,1200,2400,4800):" +
- "UNICAST(timeout=600,1200,2400,4800):" +
- "pbcast.STABLE(desired_avg_gossip=20000):" +
- "FRAG(frag_size=8192):" +
- "pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;" +
- "shun=false;print_local_addr=true):" +
- "pbcast.STATE_TRANSFER";
+ "mcast_send_buf_size=150000;mcast_recv_buf_size=80000):" +
+ "PING(timeout=1000;num_initial_members=2):" +
+ "MERGE2(min_interval=5000;max_interval=10000):" +
+ "FD_SOCK:" +
+ "VERIFY_SUSPECT(timeout=1500):" +
+ "pbcast.NAKACK(gc_lag=50;retransmit_timeout=600,1200,2400,4800):" +
+ "UNICAST(timeout=600,1200,2400,4800):" +
+ "pbcast.STABLE(desired_avg_gossip=20000):" +
+ "FRAG(frag_size=8192):" +
+ "pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;" +
+ "shun=false;print_local_addr=true):" +
+ "pbcast.STATE_TRANSFER";
}
private void initialiseCacheLoaderManager() throws CacheException
@@ -4287,7 +4293,7 @@
if (log.isDebugEnabled())
{
log.debug("Created Multiplexer Channel for cache cluster " + configuration.getClusterName() +
- " using stack " + configuration.getMultiplexerStack());
+ " using stack " + configuration.getMultiplexerStack());
}
}
else
Modified: core/trunk/src/main/java/org/jboss/cache/RPCManager.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/RPCManager.java 2007-09-24 20:54:55 UTC (rev 4501)
+++ core/trunk/src/main/java/org/jboss/cache/RPCManager.java 2007-09-25 13:05:59 UTC (rev 4502)
@@ -2,12 +2,19 @@
import org.jboss.cache.marshall.MethodCall;
import org.jgroups.Address;
+import org.jgroups.blocks.RspFilter;
import java.lang.reflect.Method;
import java.util.List;
public interface RPCManager
{
+ /**
+ * The same as {@link #callRemoteMethods(java.util.List,org.jboss.cache.marshall.MethodCall,int,boolean,long)} except that it adds a JGroups
+ * {@link org.jgroups.blocks.RspFilter} to the list of parameters, which is used to filter results.
+ */
+ public List callRemoteMethods(List<Address> recipients, MethodCall methodCall, int mode, boolean excludeSelf, long timeout, RspFilter responseFilter) throws Exception;
+
public List callRemoteMethods(List<Address> recipients, MethodCall methodCall, int mode, boolean excludeSelf, long timeout) throws Exception;
public boolean isCoordinator();
Modified: core/trunk/src/main/java/org/jboss/cache/RPCManagerImpl.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/RPCManagerImpl.java 2007-09-24 20:54:55 UTC (rev 4501)
+++ core/trunk/src/main/java/org/jboss/cache/RPCManagerImpl.java 2007-09-25 13:05:59 UTC (rev 4502)
@@ -8,6 +8,7 @@
import org.jboss.cache.marshall.MethodCall;
import org.jgroups.Address;
+import org.jgroups.blocks.RspFilter;
import java.lang.reflect.Method;
import java.util.List;
@@ -36,6 +37,12 @@
// for now, we delegate RPC calls to deprecated methods in CacheImpl.
@SuppressWarnings("deprecation")
+ public List callRemoteMethods(List<Address> recipients, MethodCall methodCall, int mode, boolean excludeSelf, long timeout, RspFilter responseFilter) throws Exception
+ {
+ return c.callRemoteMethods(recipients, methodCall, mode, excludeSelf, timeout, responseFilter);
+ }
+
+ @SuppressWarnings("deprecation")
public List callRemoteMethods(List<Address> recipients, MethodCall methodCall, int mode, boolean excludeSelf, long timeout) throws Exception
{
return c.callRemoteMethods(recipients, methodCall, mode, excludeSelf, timeout);
Modified: core/trunk/src/main/java/org/jboss/cache/loader/ClusteredCacheLoader.java
===================================================================
--- core/trunk/src/main/java/org/jboss/cache/loader/ClusteredCacheLoader.java 2007-09-24 20:54:55 UTC (rev 4501)
+++ core/trunk/src/main/java/org/jboss/cache/loader/ClusteredCacheLoader.java 2007-09-25 13:05:59 UTC (rev 4502)
@@ -20,6 +20,7 @@
import org.jboss.cache.marshall.MethodDeclarations;
import org.jgroups.Address;
import org.jgroups.blocks.GroupRequest;
+import org.jgroups.blocks.RspFilter;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
@@ -87,7 +88,10 @@
if (log.isTraceEnabled()) log.trace("cache=" + cache.getLocalAddress() + "; calling with " + call);
List<Address> mbrs = cache.getMembers();
MethodCall clusteredGet = MethodCallFactory.create(MethodDeclarations.clusteredGetMethod, call, false);
- List resps = cache.getRPCManager().callRemoteMethods(mbrs, clusteredGet, GroupRequest.GET_FIRST, true, config.getTimeout());
+ List resps = null;
+ // JBCACHE-1186
+ resps = cache.getRPCManager().callRemoteMethods(mbrs, clusteredGet, GroupRequest.GET_FIRST, true, config.getTimeout(), new ResponseValidityFilter());
+
if (resps == null)
{
if (log.isInfoEnabled()) log.info("No replies to call " + call + ". Perhaps we're alone in the cluster?");
@@ -319,4 +323,25 @@
public void setRegionManager(RegionManager manager)
{
}
+
+ public static class ResponseValidityFilter implements RspFilter
+ {
+ private int numValidResponses = 0;
+
+ public boolean isAcceptable(Object object, Address address)
+ {
+ if (!(object instanceof List)) return false;
+
+ List response = (List) object;
+ Boolean foundResult = (Boolean) response.get(0);
+ if (foundResult) numValidResponses++;
+
+ return foundResult;
+ }
+
+ public boolean needMoreResponses()
+ {
+ return numValidResponses < 1;
+ }
+ }
}
Modified: core/trunk/src/test/java/org/jboss/cache/loader/ClusteredCacheLoaderTest.java
===================================================================
--- core/trunk/src/test/java/org/jboss/cache/loader/ClusteredCacheLoaderTest.java 2007-09-24 20:54:55 UTC (rev 4501)
+++ core/trunk/src/test/java/org/jboss/cache/loader/ClusteredCacheLoaderTest.java 2007-09-25 13:05:59 UTC (rev 4502)
@@ -6,16 +6,7 @@
*/
package org.jboss.cache.loader;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-import java.util.Set;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.CountDownLatch;
-
import junit.framework.Assert;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.cache.CacheImpl;
@@ -26,12 +17,22 @@
import org.jboss.cache.lock.TimeoutException;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.CountDownLatch;
+
/**
* Tests ClusteredCacheLoader
*
* @author <a href="mailto:manik at jboss.org">Manik Surtani (manik at jboss.org)</a>
*/
+ at Test(groups = {"functional"})
public class ClusteredCacheLoaderTest extends AbstractCacheLoaderTestBase
{
private static Log log = LogFactory.getLog(ClusteredCacheLoaderTest.class);
More information about the jbosscache-commits
mailing list