HHH-10344 Make transactional invalidation backwards compatible
This commit is contained in:
Galder Zamarreño
Steve Ebersole
parent
828a83dcbf
commit
05aaeb1963
|
|
@ -58,7 +58,7 @@ test {
|
|||
// Use Infinispan's test JGroups stack that uses TEST_PING
|
||||
systemProperties['hibernate.cache.infinispan.jgroups_cfg'] = '2lc-test-tcp.xml'
|
||||
// systemProperties['log4j.configuration'] = 'file:/log4j/log4j-infinispan.xml'
|
||||
enabled = false
|
||||
enabled = project.hasProperty('testInfinispan')
|
||||
// Without this I have trouble running specific test using --tests switch
|
||||
doFirst {
|
||||
filter.includePatterns.each {
|
||||
|
|
|
|||
|
|
@ -178,6 +178,11 @@ public class PutFromLoadValidator {
|
|||
}
|
||||
boolean transactional = cache.getCacheConfiguration().transaction().transactionMode().isTransactional();
|
||||
if (transactional) {
|
||||
cache.removeInterceptor(invalidationPosition);
|
||||
TxInvalidationInterceptor txInvalidationInterceptor = new TxInvalidationInterceptor();
|
||||
cache.getComponentRegistry().registerComponent(txInvalidationInterceptor, TxInvalidationInterceptor.class);
|
||||
cache.addInterceptor(txInvalidationInterceptor, invalidationPosition);
|
||||
|
||||
// Note that invalidation does *NOT* acquire locks; therefore, we have to start invalidating before
|
||||
// wrapping the entry, since if putFromLoad was invoked between wrap and beginInvalidatingKey, the invalidation
|
||||
// would not commit the entry removal (as during wrap the entry was not in cache)
|
||||
|
|
@ -223,6 +228,13 @@ public class PutFromLoadValidator {
|
|||
cache.removeInterceptor(NonTxInvalidationInterceptor.class);
|
||||
break;
|
||||
}
|
||||
else if (i instanceof TxInvalidationInterceptor) {
|
||||
InvalidationInterceptor invalidationInterceptor = new InvalidationInterceptor();
|
||||
cache.getComponentRegistry().registerComponent(invalidationInterceptor, InvalidationInterceptor.class);
|
||||
cache.addInterceptorBefore(invalidationInterceptor, TxInvalidationInterceptor.class);
|
||||
cache.removeInterceptor(TxInvalidationInterceptor.class);
|
||||
break;
|
||||
}
|
||||
}
|
||||
CacheCommandInitializer cci = cache.getComponentRegistry().getComponent(CacheCommandInitializer.class);
|
||||
cci.removePutFromLoadValidator(cache.getName());
|
||||
|
|
|
|||
|
|
@ -0,0 +1,300 @@
|
|||
/*
|
||||
* Hibernate, Relational Persistence for Idiomatic Java
|
||||
*
|
||||
* License: GNU Lesser General Public License (LGPL), version 2.1 or later.
|
||||
* See the lgpl.txt file in the root directory or <http://www.gnu.org/licenses/lgpl-2.1.html>.
|
||||
*/
|
||||
package org.hibernate.cache.infinispan.access;
|
||||
|
||||
import org.infinispan.commands.AbstractVisitor;
|
||||
import org.infinispan.commands.CommandsFactory;
|
||||
import org.infinispan.commands.FlagAffectedCommand;
|
||||
import org.infinispan.commands.ReplicableCommand;
|
||||
import org.infinispan.commands.control.LockControlCommand;
|
||||
import org.infinispan.commands.tx.PrepareCommand;
|
||||
import org.infinispan.commands.write.ClearCommand;
|
||||
import org.infinispan.commands.write.InvalidateCommand;
|
||||
import org.infinispan.commands.write.PutKeyValueCommand;
|
||||
import org.infinispan.commands.write.PutMapCommand;
|
||||
import org.infinispan.commands.write.RemoveCommand;
|
||||
import org.infinispan.commands.write.ReplaceCommand;
|
||||
import org.infinispan.commands.write.WriteCommand;
|
||||
import org.infinispan.commons.util.InfinispanCollections;
|
||||
import org.infinispan.context.Flag;
|
||||
import org.infinispan.context.InvocationContext;
|
||||
import org.infinispan.context.impl.LocalTxInvocationContext;
|
||||
import org.infinispan.context.impl.TxInvocationContext;
|
||||
import org.infinispan.factories.annotations.Inject;
|
||||
import org.infinispan.factories.annotations.Start;
|
||||
import org.infinispan.interceptors.base.BaseRpcInterceptor;
|
||||
import org.infinispan.jmx.JmxStatisticsExposer;
|
||||
import org.infinispan.jmx.annotations.DataType;
|
||||
import org.infinispan.jmx.annotations.MBean;
|
||||
import org.infinispan.jmx.annotations.ManagedAttribute;
|
||||
import org.infinispan.jmx.annotations.ManagedOperation;
|
||||
import org.infinispan.jmx.annotations.MeasurementType;
|
||||
import org.infinispan.jmx.annotations.Parameter;
|
||||
import org.infinispan.util.logging.Log;
|
||||
import org.infinispan.util.logging.LogFactory;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
/**
|
||||
* This interceptor acts as a replacement to the replication interceptor when the CacheImpl is configured with
|
||||
* ClusteredSyncMode as INVALIDATE.
|
||||
* <p/>
|
||||
* The idea is that rather than replicating changes to all caches in a cluster when write methods are called, simply
|
||||
* broadcast an {@link InvalidateCommand} on the remote caches containing all keys modified. This allows the remote
|
||||
* cache to look up the value in a shared cache loader which would have been updated with the changes.
|
||||
*
|
||||
* @author Manik Surtani
|
||||
* @author Galder Zamarreño
|
||||
* @author Mircea.Markus@jboss.com
|
||||
* @since 4.0
|
||||
*/
|
||||
@MBean(objectName = "Invalidation", description = "Component responsible for invalidating entries on remote caches when entries are written to locally.")
|
||||
public class TxInvalidationInterceptor extends BaseRpcInterceptor implements JmxStatisticsExposer {
|
||||
|
||||
private final AtomicLong invalidations = new AtomicLong( 0 );
|
||||
private CommandsFactory commandsFactory;
|
||||
private boolean statisticsEnabled;
|
||||
|
||||
private static final Log log = LogFactory.getLog( TxInvalidationInterceptor.class );
|
||||
|
||||
@Override
|
||||
protected Log getLog() {
|
||||
return log;
|
||||
}
|
||||
|
||||
@Inject
|
||||
public void injectDependencies(CommandsFactory commandsFactory) {
|
||||
this.commandsFactory = commandsFactory;
|
||||
}
|
||||
|
||||
@Start
|
||||
private void start() {
|
||||
this.setStatisticsEnabled( cacheConfiguration.jmxStatistics().enabled() );
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable {
|
||||
if ( !isPutForExternalRead( command ) ) {
|
||||
return handleInvalidate( ctx, command, command.getKey() );
|
||||
}
|
||||
return invokeNextInterceptor( ctx, command );
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitReplaceCommand(InvocationContext ctx, ReplaceCommand command) throws Throwable {
|
||||
return handleInvalidate( ctx, command, command.getKey() );
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitRemoveCommand(InvocationContext ctx, RemoveCommand command) throws Throwable {
|
||||
return handleInvalidate( ctx, command, command.getKey() );
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitClearCommand(InvocationContext ctx, ClearCommand command) throws Throwable {
|
||||
Object retval = invokeNextInterceptor( ctx, command );
|
||||
if ( !isLocalModeForced( command ) ) {
|
||||
// just broadcast the clear command - this is simplest!
|
||||
if ( ctx.isOriginLocal() ) {
|
||||
rpcManager.invokeRemotely( null, command, rpcManager.getDefaultRpcOptions( defaultSynchronous ) );
|
||||
}
|
||||
}
|
||||
return retval;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) throws Throwable {
|
||||
Object[] keys = command.getMap() == null ? null : command.getMap().keySet().toArray();
|
||||
return handleInvalidate( ctx, command, keys );
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitPrepareCommand(TxInvocationContext ctx, PrepareCommand command) throws Throwable {
|
||||
Object retval = invokeNextInterceptor( ctx, command );
|
||||
log.tracef( "Entering InvalidationInterceptor's prepare phase. Ctx flags are empty" );
|
||||
// fetch the modifications before the transaction is committed (and thus removed from the txTable)
|
||||
if ( shouldInvokeRemoteTxCommand( ctx ) ) {
|
||||
if ( ctx.getTransaction() == null ) {
|
||||
throw new IllegalStateException( "We must have an associated transaction" );
|
||||
}
|
||||
|
||||
List<WriteCommand> mods = Arrays.asList( command.getModifications() );
|
||||
broadcastInvalidateForPrepare( mods, ctx );
|
||||
}
|
||||
else {
|
||||
log.tracef( "Nothing to invalidate - no modifications in the transaction." );
|
||||
}
|
||||
return retval;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitLockControlCommand(TxInvocationContext ctx, LockControlCommand command) throws Throwable {
|
||||
Object retVal = invokeNextInterceptor( ctx, command );
|
||||
if ( ctx.isOriginLocal() ) {
|
||||
//unlock will happen async as it is a best effort
|
||||
boolean sync = !command.isUnlock();
|
||||
( (LocalTxInvocationContext) ctx ).remoteLocksAcquired( rpcManager.getTransport().getMembers() );
|
||||
rpcManager.invokeRemotely( null, command, rpcManager.getDefaultRpcOptions( sync ) );
|
||||
}
|
||||
return retVal;
|
||||
}
|
||||
|
||||
private Object handleInvalidate(InvocationContext ctx, WriteCommand command, Object... keys) throws Throwable {
|
||||
Object retval = invokeNextInterceptor( ctx, command );
|
||||
if ( command.isSuccessful() && !ctx.isInTxScope() ) {
|
||||
if ( keys != null && keys.length != 0 ) {
|
||||
if ( !isLocalModeForced( command ) ) {
|
||||
invalidateAcrossCluster( isSynchronous( command ), keys, ctx );
|
||||
}
|
||||
}
|
||||
}
|
||||
return retval;
|
||||
}
|
||||
|
||||
private void broadcastInvalidateForPrepare(List<WriteCommand> modifications, InvocationContext ctx) throws Throwable {
|
||||
// A prepare does not carry flags, so skip checking whether is local or not
|
||||
if ( ctx.isInTxScope() ) {
|
||||
if ( modifications.isEmpty() ) {
|
||||
return;
|
||||
}
|
||||
|
||||
InvalidationFilterVisitor filterVisitor = new InvalidationFilterVisitor( modifications.size() );
|
||||
filterVisitor.visitCollection( null, modifications );
|
||||
|
||||
if ( filterVisitor.containsPutForExternalRead ) {
|
||||
log.debug( "Modification list contains a putForExternalRead operation. Not invalidating." );
|
||||
}
|
||||
else if ( filterVisitor.containsLocalModeFlag ) {
|
||||
log.debug( "Modification list contains a local mode flagged operation. Not invalidating." );
|
||||
}
|
||||
else {
|
||||
try {
|
||||
invalidateAcrossCluster( defaultSynchronous, filterVisitor.result.toArray(), ctx );
|
||||
}
|
||||
catch (Throwable t) {
|
||||
log.unableToRollbackEvictionsDuringPrepare( t );
|
||||
if ( t instanceof RuntimeException ) {
|
||||
throw (RuntimeException) t;
|
||||
}
|
||||
else {
|
||||
throw new RuntimeException( "Unable to broadcast invalidation messages", t );
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static class InvalidationFilterVisitor extends AbstractVisitor {
|
||||
|
||||
Set<Object> result;
|
||||
public boolean containsPutForExternalRead = false;
|
||||
public boolean containsLocalModeFlag = false;
|
||||
|
||||
public InvalidationFilterVisitor(int maxSetSize) {
|
||||
result = new HashSet<Object>( maxSetSize );
|
||||
}
|
||||
|
||||
private void processCommand(FlagAffectedCommand command) {
|
||||
containsLocalModeFlag = containsLocalModeFlag || ( command.getFlags() != null && command.getFlags().contains( Flag.CACHE_MODE_LOCAL ) );
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable {
|
||||
processCommand( command );
|
||||
containsPutForExternalRead =
|
||||
containsPutForExternalRead || ( command.getFlags() != null && command.getFlags().contains( Flag.PUT_FOR_EXTERNAL_READ ) );
|
||||
result.add( command.getKey() );
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitRemoveCommand(InvocationContext ctx, RemoveCommand command) throws Throwable {
|
||||
processCommand( command );
|
||||
result.add( command.getKey() );
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) throws Throwable {
|
||||
processCommand( command );
|
||||
result.addAll( command.getAffectedKeys() );
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private void invalidateAcrossCluster(boolean synchronous, Object[] keys, InvocationContext ctx) throws Throwable {
|
||||
// increment invalidations counter if statistics maintained
|
||||
incrementInvalidations();
|
||||
final InvalidateCommand invalidateCommand = commandsFactory.buildInvalidateCommand( InfinispanCollections.<Flag>emptySet(), keys );
|
||||
if ( log.isDebugEnabled() ) {
|
||||
log.debug( "Cache [" + rpcManager.getAddress() + "] replicating " + invalidateCommand );
|
||||
}
|
||||
|
||||
ReplicableCommand command = invalidateCommand;
|
||||
if ( ctx.isInTxScope() ) {
|
||||
TxInvocationContext txCtx = (TxInvocationContext) ctx;
|
||||
// A Prepare command containing the invalidation command in its 'modifications' list is sent to the remote nodes
|
||||
// so that the invalidation is executed in the same transaction and locks can be acquired and released properly.
|
||||
// This is 1PC on purpose, as an optimisation, even if the current TX is 2PC.
|
||||
// If the cache uses 2PC it's possible that the remotes will commit the invalidation and the originator rolls back,
|
||||
// but this does not impact consistency and the speed benefit is worth it.
|
||||
command = commandsFactory.buildPrepareCommand( txCtx.getGlobalTransaction(), Collections.<WriteCommand>singletonList( invalidateCommand ), true );
|
||||
}
|
||||
rpcManager.invokeRemotely( null, command, rpcManager.getDefaultRpcOptions( synchronous ) );
|
||||
}
|
||||
|
||||
private void incrementInvalidations() {
|
||||
if ( statisticsEnabled ) {
|
||||
invalidations.incrementAndGet();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isPutForExternalRead(FlagAffectedCommand command) {
|
||||
if ( command.hasFlag( Flag.PUT_FOR_EXTERNAL_READ ) ) {
|
||||
log.trace( "Put for external read called. Suppressing clustered invalidation." );
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
@ManagedOperation(
|
||||
description = "Resets statistics gathered by this component",
|
||||
displayName = "Reset statistics"
|
||||
)
|
||||
public void resetStatistics() {
|
||||
invalidations.set( 0 );
|
||||
}
|
||||
|
||||
@ManagedAttribute(
|
||||
displayName = "Statistics enabled",
|
||||
description = "Enables or disables the gathering of statistics by this component",
|
||||
dataType = DataType.TRAIT,
|
||||
writable = true
|
||||
)
|
||||
public boolean getStatisticsEnabled() {
|
||||
return this.statisticsEnabled;
|
||||
}
|
||||
|
||||
public void setStatisticsEnabled(
|
||||
@Parameter(name = "enabled", description = "Whether statistics should be enabled or disabled (true/false)") boolean enabled) {
|
||||
this.statisticsEnabled = enabled;
|
||||
}
|
||||
|
||||
@ManagedAttribute(
|
||||
description = "Number of invalidations",
|
||||
displayName = "Number of invalidations",
|
||||
measurementType = MeasurementType.TRENDSUP
|
||||
)
|
||||
public long getInvalidations() {
|
||||
return invalidations.get();
|
||||
}
|
||||
}
|
||||
|
|
@ -84,11 +84,18 @@ class TxPutFromLoadInterceptor extends BaseRpcInterceptor {
|
|||
if (wc instanceof InvalidateCommand) {
|
||||
// ISPN-5605 InvalidateCommand does not correctly implement getAffectedKeys()
|
||||
for (Object key : ((InvalidateCommand) wc).getKeys()) {
|
||||
if (log.isTraceEnabled()) {
|
||||
log.tracef("Invalidating key %s with lock owner %s", key, ctx.getLockOwner());
|
||||
}
|
||||
putFromLoadValidator.beginInvalidatingKey(ctx.getLockOwner(), key);
|
||||
}
|
||||
}
|
||||
else {
|
||||
for (Object key : wc.getAffectedKeys()) {
|
||||
Set<Object> keys = wc.getAffectedKeys();
|
||||
if (log.isTraceEnabled()) {
|
||||
log.tracef("Invalidating keys %s with lock owner %s", keys, ctx.getLockOwner());
|
||||
}
|
||||
for (Object key : keys ) {
|
||||
putFromLoadValidator.beginInvalidatingKey(ctx.getLockOwner(), key);
|
||||
}
|
||||
}
|
||||
|
|
@ -99,11 +106,19 @@ class TxPutFromLoadInterceptor extends BaseRpcInterceptor {
|
|||
|
||||
@Override
|
||||
public Object visitCommitCommand(TxInvocationContext ctx, CommitCommand command) throws Throwable {
|
||||
if (log.isTraceEnabled()) {
|
||||
log.tracef( "Commit command received, end invalidation" );
|
||||
}
|
||||
|
||||
return endInvalidationAndInvokeNextInterceptor(ctx, command);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object visitRollbackCommand(TxInvocationContext ctx, RollbackCommand command) throws Throwable {
|
||||
if (log.isTraceEnabled()) {
|
||||
log.tracef( "Rollback command received, end invalidation" );
|
||||
}
|
||||
|
||||
return endInvalidationAndInvokeNextInterceptor(ctx, command);
|
||||
}
|
||||
|
||||
|
|
@ -112,6 +127,11 @@ class TxPutFromLoadInterceptor extends BaseRpcInterceptor {
|
|||
if (ctx.isOriginLocal()) {
|
||||
// send async Commit
|
||||
Set<Object> affectedKeys = ctx.getAffectedKeys();
|
||||
|
||||
if (log.isTraceEnabled()) {
|
||||
log.tracef( "Sending end invalidation for keys %s asynchronously", affectedKeys );
|
||||
}
|
||||
|
||||
if (!affectedKeys.isEmpty()) {
|
||||
EndInvalidationCommand commitCommand = cacheCommandInitializer.buildEndInvalidationCommand(
|
||||
cacheName, affectedKeys.toArray(), ctx.getGlobalTransaction());
|
||||
|
|
|
|||
|
|
@ -47,6 +47,7 @@ import java.util.concurrent.CountDownLatch;
|
|||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
|
|
@ -334,7 +335,7 @@ public abstract class AbstractRegionAccessStrategyTest<R extends BaseRegion, S e
|
|||
|
||||
// Test whether the get above messes up the optimistic version
|
||||
SessionImplementor s9 = mockedSession();
|
||||
remoteAccessStrategy.putFromLoad(s9, KEY, VALUE1, s9.getTimestamp(), 1);
|
||||
assertTrue(remoteAccessStrategy.putFromLoad(s9, KEY, VALUE1, s9.getTimestamp(), 1));
|
||||
SessionImplementor s10 = mockedSession();
|
||||
assertEquals(VALUE1, remoteAccessStrategy.get(s10, KEY, s10.getTimestamp()));
|
||||
assertEquals(1, remoteRegion.getCache().size());
|
||||
|
|
|
|||
Loading…
Reference in New Issue