[jbosscache-commits] JBoss Cache SVN: r5584 - in core/trunk/src: main/java/org/jboss/cache/commands/cachedata and 4 other directories.

Author: manik.surtani(a)jboss.com Date: 2008-04-17 14:07:17 -0400 (Thu, 17 Apr 2008) New Revision: 5584 Modified: core/trunk/src/main/java/org/jboss/cache/buddyreplication/BuddyManager.java core/trunk/src/main/java/org/jboss/cache/commands/cachedata/GetKeyValueCommand.java core/trunk/src/main/java/org/jboss/cache/commands/cachedata/PutDataMapCommand.java core/trunk/src/main/java/org/jboss/cache/commands/remote/DataGravitationCleanupCommand.java core/trunk/src/main/java/org/jboss/cache/interceptors/BaseRpcInterceptor.java core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java core/trunk/src/main/java/org/jboss/cache/invocation/CacheInvocationDelegate.java core/trunk/src/test/java/org/jboss/cache/buddyreplication/BuddyAssignmentStateTransferTest.java Log: Updated data gravitation code Modified: core/trunk/src/main/java/org/jboss/cache/buddyreplication/BuddyManager.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/buddyreplication/BuddyManager.java 2008-04-17 16:32:43 UTC (rev 5583) +++ core/trunk/src/main/java/org/jboss/cache/buddyreplication/BuddyManager.java 2008-04-17 18:07:17 UTC (rev 5584) @@ -558,7 +558,6 @@ // create the backup region anyway Option o = cache.getInvocationContext().getOptionOverrides(); o.setSkipCacheStatusCheck(true); - Node root = cache.getRoot(); o = cache.getInvocationContext().getOptionOverrides(); o.setCacheModeLocal(true); o.setSkipCacheStatusCheck(true); @@ -574,6 +573,7 @@ Fqn fqn = entry.getKey(); if (!regionManager.isInactive(fqn)) { + if (trace) log.trace("Integrating state for region " + fqn); //ClassLoader cl = (marshaller == null) ? null : marshaller.getClassLoader(fqnS); Fqn integrationRoot = Fqn.fromRelativeFqn(integrationBase, fqn); @@ -609,6 +609,10 @@ } } } + else + { + log.trace("Received state for region " + fqn + " but region is inactive"); + } } } } @@ -819,6 +823,10 @@ } } } + else + { + if (trace) log.trace("Not configured to provide state!"); + } // now broadcast a message to the newly assigned buddies. AssignToBuddyGroupCommand membershipCall = commandsFactory.buildAssignToBuddyGroupCommand(buddyGroup, stateMap); Modified: core/trunk/src/main/java/org/jboss/cache/commands/cachedata/GetKeyValueCommand.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/commands/cachedata/GetKeyValueCommand.java 2008-04-17 16:32:43 UTC (rev 5583) +++ core/trunk/src/main/java/org/jboss/cache/commands/cachedata/GetKeyValueCommand.java 2008-04-17 18:07:17 UTC (rev 5584) @@ -111,4 +111,14 @@ result = 31 * result + (sendNodeEvent ? 1 : 0); return result; } + + + public String toString() + { + return "GetKeyValueCommand{" + + "fqn=" + fqn + + ", key=" + key + + ", sendNodeEvent=" + sendNodeEvent + + '}'; + } } Modified: core/trunk/src/main/java/org/jboss/cache/commands/cachedata/PutDataMapCommand.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/commands/cachedata/PutDataMapCommand.java 2008-04-17 16:32:43 UTC (rev 5583) +++ core/trunk/src/main/java/org/jboss/cache/commands/cachedata/PutDataMapCommand.java 2008-04-17 18:07:17 UTC (rev 5584) @@ -181,7 +181,8 @@ public String toString() { return "PutDataMapCommand{" + - "eraseContents=" + eraseContents + + "fqn=" + fqn + + ", eraseContents=" + eraseContents + ", createUndoOps=" + createUndoOps + ", dataVersion=" + dataVersion + ", data=" + data + Modified: core/trunk/src/main/java/org/jboss/cache/commands/remote/DataGravitationCleanupCommand.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/commands/remote/DataGravitationCleanupCommand.java 2008-04-17 16:32:43 UTC (rev 5583) +++ core/trunk/src/main/java/org/jboss/cache/commands/remote/DataGravitationCleanupCommand.java 2008-04-17 18:07:17 UTC (rev 5584) @@ -27,7 +27,7 @@ * todo this calls other commands through the invocation chain, check wheter direct call is not okay * todo it is not natural for this command to extend TxCC, revisit */ -public class DataGravitationCleanupCommand extends BaseCacheDataCommand implements TxCacheCommand, GlobalTransactionCommand +public class DataGravitationCleanupCommand extends BaseCacheDataCommand implements TxCacheCommand, GlobalTransactionCommand, DirectCommand { public static final int METHOD_ID = 34; @@ -62,8 +62,13 @@ this.commandsFactory = commandsFactory; } - public Object perform(InvocationContext ctx) throws Throwable + public Object perform(InvocationContext ctx) { + throw new UnsupportedOperationException("Direct commands are not meant to be passed up the interceptor chain!"); + } + + public Object performDirectly() throws Throwable + { if (buddyManager.isDataGravitationRemoveOnFind()) { if (trace) Modified: core/trunk/src/main/java/org/jboss/cache/interceptors/BaseRpcInterceptor.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/interceptors/BaseRpcInterceptor.java 2008-04-17 16:32:43 UTC (rev 5583) +++ core/trunk/src/main/java/org/jboss/cache/interceptors/BaseRpcInterceptor.java 2008-04-17 18:07:17 UTC (rev 5584) @@ -96,21 +96,40 @@ protected void replicateCall(InvocationContext ctx, List<Address> recipients, CacheCommand c, boolean sync, Option o, boolean wrapCacheCommandInReplicateMethod, boolean useOutOfBandMessage) throws Throwable { + long syncReplTimeout = config.getSyncReplTimeout(); + + // test for option overrides + if (o != null) + { + if (o.isForceAsynchronous()) sync = false; + else if (o.isForceSynchronous()) sync = true; + + if (o.getSyncReplTimeout() > 0) syncReplTimeout = o.getSyncReplTimeout(); + } + + // tx-level overrides are more important + Transaction tx = ctx.getTransaction(); + if (tx != null) + { + GlobalTransaction gtx = ctx.getGlobalTransaction(); + TransactionEntry te = txTable.get(gtx); + if (te != null) + { + if (te.isForceAsyncReplication()) sync = false; + else if (te.isForceSyncReplication()) sync = true; + } + } + + replicateCall(recipients, c, sync, wrapCacheCommandInReplicateMethod, useOutOfBandMessage, false, (int) syncReplTimeout); + } + + protected void replicateCall(List<Address> recipients, CacheCommand c, boolean sync, boolean wrapCacheCommandInReplicateMethod, boolean useOutOfBandMessage, boolean isBroadcast, int timeout) throws Throwable + { if (c instanceof MarshallableCommand) { MarshallableCommand call = (MarshallableCommand) c; if (trace) log.trace("Broadcasting call " + call + " to recipient list " + recipients); - Transaction tx = ctx.getTransaction(); - if (tx != null) - { - GlobalTransaction gtx = ctx.getGlobalTransaction(); - TransactionEntry te = txTable.get(gtx); - if (te != null) - { - if (te.isForceAsyncReplication()) sync = false; - else if (te.isForceSyncReplication()) sync = true; - } - } + if (!sync && replicationQueue != null && !usingBuddyReplication) { if (log.isDebugEnabled()) log.debug("Putting call " + call + " on the replication queue."); @@ -118,26 +137,23 @@ } else { - if (usingBuddyReplication) call = buddyManager.transformFqns(call); + if (usingBuddyReplication && !isBroadcast) call = buddyManager.transformFqns(call); List<Address> callRecipients = recipients; if (callRecipients == null) { - callRecipients = usingBuddyReplication ? buddyManager.getBuddyAddresses() : rpcManager.getMembers(); + callRecipients = usingBuddyReplication && !isBroadcast ? buddyManager.getBuddyAddresses() : rpcManager.getMembers(); if (trace) log.trace("Setting call recipients to " + callRecipients + " since the original list of recipients passed in is null."); } - long syncReplTimeout = o.getSyncReplTimeout(); - if (syncReplTimeout < 0) syncReplTimeout = config.getSyncReplTimeout(); - CacheCommand toCall = wrapCacheCommandInReplicateMethod ? commandsFactory.buildReplicateCommand(call) : call; List rsps = rpcManager.callRemoteMethods(callRecipients, toCall, sync, // is synchronised? true, // ignore self? - (int) syncReplTimeout, + timeout, useOutOfBandMessage ); if (trace) log.trace("responses=" + rsps); Modified: core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java 2008-04-17 16:32:43 UTC (rev 5583) +++ core/trunk/src/main/java/org/jboss/cache/interceptors/DataGravitatorInterceptor.java 2008-04-17 18:07:17 UTC (rev 5584) @@ -25,8 +25,6 @@ import org.jboss.cache.commands.remote.GravitateDataCommand; import org.jboss.cache.commands.remote.RemoteExistsNodeCommand; import org.jboss.cache.commands.tx.CommitCommand; -import org.jboss.cache.commands.tx.OptimisticPrepareCommand; -import org.jboss.cache.commands.tx.PrepareCommand; import org.jboss.cache.commands.tx.RollbackCommand; import org.jboss.cache.config.Configuration; import org.jboss.cache.factories.annotations.Inject; @@ -40,6 +38,7 @@ import org.jgroups.blocks.RspFilter; import java.util.ArrayList; +import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -56,6 +55,12 @@ * See the JBoss Cache User Guide for more details on configuration options. * There is a section dedicated to Buddy Replication in the Replication * chapter. + * <p/> + * In terms of functionality, if a gravitation call has occured and a cleanup call is needed (based on + * how BR is configured), a cleanup call will be broadcast immediately after the gravitation call (no txs) + * or if txs are used, an <i>asynchronous</i> call is made to perform the cleanup <i>outside</i> the scope + * of the tx that caused the gravitation event. + * <p/> * * @author <a href="mailto:manik@jboss.org">Manik Surtani (manik(a)jboss.org)&lt;/a&gt; */ @@ -63,12 +68,17 @@ { private BuddyManager buddyManager; private boolean syncCommunications = false; - private Map<GlobalTransaction, CacheCommand> transactionMods = new ConcurrentHashMap<GlobalTransaction, CacheCommand>(); + /** + * Map that contains commands that need cleaning up. This is keyed on global transaction, and contains a list of + * cleanup commands corresponding to all gravitate calls made during the course of the transaction in question. + */ + private Map<GlobalTransaction, List<CacheCommand>> cleanupCommands = new ConcurrentHashMap<GlobalTransaction, List<CacheCommand>>(); private CacheData cacheData; private Configuration config; private CommandsFactory commandsFactory; private CacheSPI cacheSPI; private boolean usingOptimisticLocking; + private boolean needToCleanup; @Inject public void injectComponents(BuddyManager buddyManager, Configuration config, CacheData cacheData, CommandsFactory commandsFactory, CacheSPI cacheSPI) @@ -85,6 +95,7 @@ { syncCommunications = config.getCacheMode() == Configuration.CacheMode.REPL_SYNC || config.getCacheMode() == Configuration.CacheMode.INVALIDATION_SYNC; usingOptimisticLocking = config.isNodeLockingOptimistic(); + needToCleanup = config.getBuddyReplicationConfig().isDataGravitationRemoveOnFind(); } @Override @@ -124,67 +135,30 @@ } @Override - public Object handlePrepareCommand(InvocationContext ctx, PrepareCommand command) throws Throwable - { - try - { - Object returnValue = invokeNextInterceptor(ctx, command); - doPrepare(ctx.getGlobalTransaction(), ctx); - return returnValue; - } - catch (Throwable throwable) - { - transactionMods.remove(ctx.getGlobalTransaction()); - throw throwable; - - } - } - - @Override - public Object handleOptimisticPrepareCommand(InvocationContext ctx, OptimisticPrepareCommand command) throws Throwable - { - try - { - Object returnValue = invokeNextInterceptor(ctx, command); - doPrepare(ctx.getGlobalTransaction(), ctx); - return returnValue; - } - catch (Throwable throwable) - { - transactionMods.remove(ctx.getGlobalTransaction()); - throw throwable; - - } - } - - @Override public Object handleRollbackCommand(InvocationContext ctx, RollbackCommand command) throws Throwable { try { - transactionMods.remove(ctx.getGlobalTransaction()); return invokeNextInterceptor(ctx, command); } - catch (Throwable throwable) + finally { - transactionMods.remove(ctx.getGlobalTransaction()); - throw throwable; + cleanupCommands.remove(ctx.getGlobalTransaction()); } } @Override public Object handleCommitCommand(InvocationContext ctx, CommitCommand command) throws Throwable { + GlobalTransaction gtx = ctx.getGlobalTransaction(); try { - doCommit(ctx.getGlobalTransaction(), ctx); - transactionMods.remove(ctx.getGlobalTransaction()); + doCommit(gtx); return invokeNextInterceptor(ctx, command); } - catch (Throwable throwable) + finally { - transactionMods.remove(ctx.getGlobalTransaction()); - throw throwable; + cleanupCommands.remove(gtx); } } @@ -202,6 +176,8 @@ { if (cacheData.peek(command.getFqn(), false, false) == null) { + // gravitation is necessary. + if (trace) log.trace("Gravitating from local backup tree"); BackupData data = localBackupGet(command.getFqn(), ctx); @@ -214,20 +190,16 @@ if (data != null) { - // create node locally so I don't gravitate again - // when I do the put() call to the cluster! - //createNode(data.backupData, true); - // Make sure I replicate to my buddies. if (trace) log.trace("Passing the put call locally to make sure state is persisted and ownership is correctly established."); - createNode(ctx, data.backupData, false); + // store the gravitated node locally. This will cause it being backed up in the current instance's buddy. + createNode(data.backupData); - // very strange, the invocation contexts get twisted up here, and will need preservation. - // a bit crappy and hacky, all will be solved when we move to JBoss AOP in 2.1.0 - //ctx.setMethodCall(m); - - // Clean up the other nodes - cleanBackupData(data, ctx.getGlobalTransaction(), ctx); + if (needToCleanup) + { + // Clean up the previous data owner and backups made in the name of the previous owner. + cleanBackupData(data, ctx); + } } } else @@ -259,44 +231,30 @@ return enabled; } - private void doPrepare(GlobalTransaction gtx, InvocationContext ctx) throws Throwable + private void doCommit(GlobalTransaction gtx) throws Throwable { - CacheCommand cleanup = transactionMods.get(gtx); - if (trace) log.trace("Broadcasting prepare for cleanup ops " + cleanup); - if (cleanup != null) + if (cleanupCommands.containsKey(gtx)) { - CacheCommand prepare; - if (usingOptimisticLocking) + if (trace) log.trace("Broadcasting cleanup commands for gtx " + gtx); + + for (CacheCommand command : cleanupCommands.get(gtx)) { - prepare = commandsFactory.buildOptimisticPrepareCommand(gtx, cleanup); + try + { + doCleanup(command); + } + catch (Throwable th) + { + log.warn("Problems performing gravitation cleanup. Cleanup command: " + command, th); + } } - else - { - prepare = commandsFactory.buildPrepareCommand(gtx, cleanup, !syncCommunications); - } - - replicateCall(ctx, buddyManager.getMembersOutsideBuddyGroup(), prepare, syncCommunications, ctx.getOptionOverrides()); } else { - if (trace) log.trace("Nothing to broadcast in prepare phase for gtx " + gtx); + if (trace) log.trace("No cleanups to broadcast in commit phase for gtx " + gtx); } } - private void doCommit(GlobalTransaction gtx, InvocationContext ctx) throws Throwable - { - if (transactionMods.containsKey(gtx)) - { - if (trace) log.trace("Broadcasting commit for gtx " + gtx); - CommitCommand call = commandsFactory.buildCommitCommand(gtx); - replicateCall(ctx, buddyManager.getMembersOutsideBuddyGroup(), call, syncCommunications, ctx.getOptionOverrides(), true, true); - } - else - { - if (trace) log.trace("Nothing to broadcast in commit phase for gtx " + gtx); - } - } - private BackupData remoteBackupGet(Fqn name) throws Exception { BackupData result = null; @@ -313,32 +271,46 @@ return result; } - private void cleanBackupData(BackupData backup, GlobalTransaction gtx, InvocationContext ctx) throws Throwable + private void cleanBackupData(BackupData backup, InvocationContext ctx) throws Throwable { DataGravitationCleanupCommand cleanupCommand = commandsFactory.buildDataGravitationCleanupCommand(backup.primaryFqn, backup.backupFqn); + GlobalTransaction gtx = ctx.getGlobalTransaction(); - if (trace) log.trace("Performing cleanup on [" + backup.primaryFqn + "]"); if (gtx == null) { // broadcast removes // remove main Fqn - if (trace) log.trace("Performing cleanup on [" + backup.backupFqn + "]"); + if (trace) log.trace("Performing cleanup on [" + backup.primaryFqn + "]"); // remove backup Fqn - replicateCall(ctx, rpcManager.getMembers(), cleanupCommand, syncCommunications, ctx.getOptionOverrides(), false, false); + doCleanup(cleanupCommand); } else { if (trace) + log.trace("Data gravitation performed under global transaction " + gtx + ". Not broadcasting cleanups until the tx commits. Recording cleanup command for later use."); + List<CacheCommand> commands; + if (cleanupCommands.containsKey(gtx)) { - log.trace("Data gravitation performed under global transaction " + gtx + ". Not broadcasting cleanups until the tx commits. Adding to tx mod list instead."); + commands = cleanupCommands.get(gtx); } - transactionMods.put(gtx, cleanupCommand); - TransactionEntry te = getTransactionEntry(gtx); - te.addModification(cleanupCommand); + else + { + commands = new LinkedList<CacheCommand>(); + } + + commands.add(cleanupCommand); + cleanupCommands.put(gtx, commands); } } + private void doCleanup(CacheCommand cleanupCommand) throws Throwable + { + // cleanup commands are always ASYNCHRONOUS and is broadcast to *everyone* (even members of the current buddy + // group as they may be members of > 1 buddy group) + replicateCall(null, cleanupCommand, false, false, false, true, -1); + } + private GravitateResult gravitateData(Fqn fqn) throws Exception { if (trace) log.trace("Requesting data gravitation for Fqn " + fqn); @@ -391,22 +363,11 @@ } @SuppressWarnings("unchecked") - private void createNode(InvocationContext ctx, List<NodeData> nodeData, boolean localOnly) throws CacheException + private void createNode(List<NodeData> nodeData) throws CacheException { for (NodeData data : nodeData) { - if (localOnly) - { -// if (cache.peek(data.getFqn(), false) == null) - if (cacheData.peek(data.getFqn(), false, false) == null) - { - cacheData.createNodesLocally((Fqn<?>) data.getFqn(), (Map<?, ?>) data.getAttributes()); - } - } - else - { - cacheSPI.put(data.getFqn(), data.getAttributes()); - } + cacheSPI.put(data.getFqn(), data.getAttributes()); } } Modified: core/trunk/src/main/java/org/jboss/cache/invocation/CacheInvocationDelegate.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/invocation/CacheInvocationDelegate.java 2008-04-17 16:32:43 UTC (rev 5583) +++ core/trunk/src/main/java/org/jboss/cache/invocation/CacheInvocationDelegate.java 2008-04-17 18:07:17 UTC (rev 5584) @@ -333,7 +333,6 @@ public void move(Fqn<?> nodeToMove, Fqn<?> newParent) throws NodeNotExistsException { - checkState(); MoveCommand command = commandsFactory.buildMoveCommand(nodeToMove, newParent); invoke(command); } @@ -355,7 +354,6 @@ public void evict(Fqn<?> fqn, boolean recursive) { - checkState(); List<Fqn> nodesToEvict = cacheData.getNodesForEviction(fqn, recursive); for (Fqn aFqn : nodesToEvict) { @@ -370,7 +368,6 @@ public V get(Fqn<?> fqn, K key) { - checkState(); GetKeyValueCommand command = commandsFactory.buildGetKeyValueCommand(fqn, key, true); return (V) invoke(command); } @@ -382,7 +379,6 @@ public boolean removeNode(Fqn<?> fqn) { - checkState(); // special case if we are removing the root. Remove all children instead. if (fqn.isRoot()) { @@ -410,11 +406,6 @@ } } - private void checkState() - { - if (!getCacheStatus().allowInvocations()) throw new IllegalStateException("Cache is not in STARTED state"); - } - public boolean removeNode(String fqn) { return removeNode(Fqn.fromString(fqn)); @@ -422,7 +413,6 @@ public NodeSPI<K, V> getNode(Fqn<?> fqn) { - checkState(); GetNodeCommand command = commandsFactory.buildGetNodeCommand(fqn); return (NodeSPI) invoke(command); } @@ -434,7 +424,6 @@ public V remove(Fqn<?> fqn, K key) throws CacheException { - checkState(); GlobalTransaction tx = transactionHelper.getCurrentTransaction(); RemoveKeyCommand command = commandsFactory.buildRemoveKeyCommand(tx, fqn, key, true); return (V) invoke(command); @@ -447,7 +436,6 @@ public void put(Fqn<?> fqn, Map<K, V> data) { - checkState(); PutDataMapCommand command = commandsFactory.buildPutDataMapCommand(null, fqn, data, true, false); invoke(command); } @@ -459,7 +447,6 @@ public void putForExternalRead(Fqn<?> fqn, K key, V value) { - checkState(); // if the node exists then this should be a no-op. if (peek(fqn, false, false) == null) { @@ -477,7 +464,6 @@ public V put(Fqn<?> fqn, K key, V value) { - checkState(); GlobalTransaction tx = transactionHelper.getCurrentTransaction(); PutKeyValueCommand command = commandsFactory.buildPutKeyValueCommand(tx, fqn, key, value, false, false); return (V) invoke(command); @@ -512,7 +498,6 @@ */ public Map<K, V> getData(Fqn<?> fqn) { - checkState(); GetDataMapCommand command = commandsFactory.buildGetDataMapCommand(fqn); return (Map<K, V>) invoke(command); } @@ -538,7 +523,6 @@ */ public Set<K> getKeys(Fqn<?> fqn) { - checkState(); GetKeysCommand command = commandsFactory.buildGetKeysCommand(fqn); return (Set<K>) invoke(command); } @@ -556,7 +540,6 @@ */ public void clearData(Fqn fqn) { - checkState(); GlobalTransaction tx = getCurrentTransaction(); invoke(commandsFactory.buildRemoveDataCommand(tx, fqn, true, false, false)); } Modified: core/trunk/src/test/java/org/jboss/cache/buddyreplication/BuddyAssignmentStateTransferTest.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/buddyreplication/BuddyAssignmentStateTransferTest.java 2008-04-17 16:32:43 UTC (rev 5583) +++ core/trunk/src/test/java/org/jboss/cache/buddyreplication/BuddyAssignmentStateTransferTest.java 2008-04-17 18:07:17 UTC (rev 5584) @@ -45,7 +45,7 @@ Fqn<String> test = BuddyFqnTransformer.getBackupFqn(caches.get(0).getLocalAddress(), main); - assertEquals("State transferred", "Joe", caches.get(1).get(test, "name")); + assertEquals("State not transferred", "Joe", caches.get(1).get(test, "name")); caches.add(createCache(1, "TEST", false, true));