ISPN-7961 Cross-site replication of functional commands #5232

rvansa · 2017-06-26T12:54:49Z

Fix handling of ComputeCommand and ComputeIfAbsentCommand
In transaction replicate values from transactional context (replaying
modifications in remote site may have different results)

https://issues.jboss.org/browse/ISPN-7961

pruivo · 2017-06-30T09:24:28Z

@rvansa needs rebase

pruivo

initial review

pruivo · 2017-06-30T09:37:59Z

core/src/test/java/org/infinispan/xsite/AbstractTwoSitesTest.java

@@ -122,7 +131,36 @@ protected String key(String site) {
   }


+   public CacheOperationsTest cacheMode(CacheMode cacheMode) {


wrong return type in cacheMode(), transactional() and lockingMode() methods.

pruivo · 2017-06-30T09:43:03Z

core/src/main/java/org/infinispan/xsite/BackupSenderImpl.java

+      // All we can do is to assume that if the user plays with unsafe flags he won't modify the entry once
+      // in a replicable and another time in a non-replicable way
+      //TODO: what if after functional command the modified entry is not in the context?
+      Map<Object, WriteCommand> lastModifyingCommand = new HashMap<>();
      for (WriteCommand writeCommand : modifications) {


This won't work if you have assertions enabled. If you have a transaction with put, remove, put on the same key, when iterating over the remove, the lastModification won't be a RemoveCommand.

I would reverse iterate over the modification (can be changed to a List) and if the key exists in lastModifyingCommand map, it would continue for the next key/command.

Reverse iteration is much better, done.

pruivo · 2017-06-30T09:53:34Z

core/src/main/java/org/infinispan/interceptors/distribution/BaseDistributionInterceptor.java

@@ -358,44 +360,55 @@ public Object visitGetAllCommand(InvocationContext ctx, GetAllCommand command) t
         }
         return invokeNext(ctx, command);
      }
+      GetAllSuccessHandler getAllSuccessHandler = new GetAllSuccessHandler(command);


are this changes related to xsite?

Yes, I need a way to fetch multiple entries (remoteGetAll) from TxDistributionInterceptor, because when xsite is on, we will need the final value of the entry to replicate xsite.

I'm not convinced. the read commands aren't sent to the backup site.

You got me wrong; When you execute functional modification in a transaction, by default you don't fetch the whole entry into originator's context, just the version for WSC. However, currently the xsite is implemented in such a way that the whole transaction is replicated xsite by the originator. And since the xsite may not have the previous value, we can't just replay the func modification in there - we need to send the whole updated value. Therefore, if xsite is on, we need to fetch the modified entry in order to have the updated value on the originator.
Since some commands work on multiple entries, I need some kind of remoteGetAll to get hold of those values. Note that at some places the commands use just multiple remoteGets but that's a bug, in fact https://issues.jboss.org/browse/ISPN-7889

@rvansa I think @pruivo wanted to ask if the interceptor couldn't have have created called visitGetAllCommand when it needs some remote values, without changing the handling of GetAllCommand.

Calling visitGetAllCommand would also mean that it would call the interceptors below, executing the command and then checking if it loaded all entries or throwing OTE (as write commands should). I think that the error handling was the main reason why I went for the refactoring.

pruivo · 2017-06-30T10:02:23Z

core/src/main/java/org/infinispan/commands/functional/WriteOnlyKeyCommand.java

@@ -8,6 +8,7 @@
 import org.infinispan.commands.CommandInvocationId;
 import org.infinispan.commands.Visitor;
 import org.infinispan.commands.write.ValueMatcher;
+import org.infinispan.functional.EntryView;


unused import

danberindei · 2017-07-05T14:33:01Z

core/src/main/java/org/infinispan/functional/Param.java

       */
      CLUSTER,
      /**
       * Command is executed only locally, it is not sent to remote nodes. If the command is a write and this node
       * is not an owner of given entry, the entry is not stored in the cache. If the command reads a value and
       * the entry is not available locally, null entry is provided instead.
       */
-      LOCAL;
+      LOCAL,


It's a bit unclear what happens with writes when the originator is a backup owner, is the entry updated without contacting the primary owner? I'll grant you that's not that clear for non-functional writes and CACHE_MODE_LOCAL, either, but we have to start somewhere :)

danberindei · 2017-07-05T14:46:23Z

core/src/main/java/org/infinispan/functional/Param.java

+       * Command is executed only in the current site (same as {@link #CLUSTER}, but it is not sent for backup
+       * to other sites)
+       */
+      SITE_ONLY;


I think it's a confusing to use SITE_ONLY when the command is only executed in the local cluster, and CLUSTER when it's also sent to the backup clusters. I'd like to rename CLUSTER to ALL (just deprecating CLUSTER for now, of course) and SITE_ONLY to LOCAL_SITE.

Also, I'd remove the details about where exactly the command is executed with CLUSTER, since we're never going to include here the entire replication algorithm, with retries and so on. And we could still decide to send the full value from the primary to the backups in non-tx caches, or from the originator to the owners in tx caches ;)

I'll rename CLUSTER to ALL, we can mess with this a bit due to the @Experimental flag. It's not used anywhere, as it's the default anyway. And LOCAL_SITE sounds good.

I'll add some backdoor to the documentation of CLUSTER :)

danberindei · 2017-07-06T06:16:05Z

core/src/main/java/org/infinispan/interceptors/distribution/BaseDistributionInterceptor.java

   }

   protected void handleRemotelyRetrievedKeys(InvocationContext ctx, List<?> remoteKeys) {
   }

-   private class ClusteredGetAllHandler implements BiConsumer<Map<Address, Response>, Throwable> {
+   private class ClusteredGetAllHandler<C extends FlagAffectedCommand & TopologyAffectedCommand> implements BiConsumer<Map<Address, Response>, Throwable> {


Isn't it time we introduced an interface that extends both FlagAffectedCommand and TopologyAffectedCommand?

Maybe, but I wouldn't add changes in command hierarchy into this PR.

danberindei · 2017-07-06T06:18:15Z

core/src/main/java/org/infinispan/interceptors/distribution/BaseDistributionInterceptor.java

@@ -358,44 +360,55 @@ public Object visitGetAllCommand(InvocationContext ctx, GetAllCommand command) t
         }
         return invokeNext(ctx, command);
      }
+      GetAllSuccessHandler getAllSuccessHandler = new GetAllSuccessHandler(command);


@rvansa I think @pruivo wanted to ask if the interceptor couldn't have have created called visitGetAllCommand when it needs some remote values, without changing the handling of GetAllCommand.

danberindei · 2017-07-06T06:27:37Z

core/src/main/java/org/infinispan/interceptors/distribution/BaseDistributionInterceptor.java

+               try {
+                  remoteGetAllHandler.onKeysLost(keys);
+               } catch (Throwable t) {
+                  allFuture.completeExceptionally(t);


I think this should be outside the synchronized block, so that late responses don't have to block while the remaining interceptor callbacks are running.

completeExceptionally must be called within a synchronized block, see ClusteredGetAllFuture javadoc. I could add another allFuture.isDone() check before the sync block on line 430, but we need to sync the responses.

I'm not sure I understand the javadoc of ClusteredGetAllFuture either...

Completing allFuture will run some interceptor callbacks synchronously, but not all of them. E.g. the state transfer interceptor callback would suspend the invocation to wait for a new topology, and that would release the allFuture monitor, but the new topology could arrive while remoteGetAllHandler.onKeysLost(keys) is running, and you'd still have 2 threads accessing the same invocation context.

So I don't think synchronizing completeExceptionally() helps, we need to cancel all the BaseDistributionInterceptor-related callbacks before completing allFuture instead.

When completeExceptionally executes before the other responses processing, the future will be marked as done and as soon as the other responses get into the synchronized block, these will check isDone and return.
If the response is being processed in sync block, running completeExceptionally and the related callbacks will be blocked until we finish the processing.

Sure, having the handlers check isDone() works. But the ClusteredGetAllFuture javadoc only mentions synchronizing around completeExceptionally(), and that's not enough.

And yes, please add an isDone() check before entering the synchronized block.

danberindei · 2017-07-06T06:44:04Z

core/src/main/java/org/infinispan/interceptors/distribution/TxDistributionInterceptor.java

+         if (forceRemoteReadForFunctionalCommands && !command.hasAnyFlag(FlagBitSets.SKIP_XSITE_BACKUP)) {
+            CompletableFuture<Void> cf = remoteGetAll(ctx, command, command.getAffectedKeys(), RemoteGetAllForWriteHandler.INSTANCE);
+            if (cf == null) {
+               return invokeNext(ctx, command);


remoteGetAll() could return a CompletableFutures.completedNull(), and then you wouldn't need the if.

danberindei · 2017-07-06T08:44:27Z

core/src/main/java/org/infinispan/xsite/BaseBackupReceiver.java

-                            metadata.lifespan(), TimeUnit.MILLISECONDS,
-                            metadata.maxIdle(), TimeUnit.MILLISECONDS);
+      public Object visitClearCommand(InvocationContext ctx, ClearCommand command) throws Throwable {
+         backupCache.clear();


Missed another TODO here ;)

danberindei · 2017-07-06T08:46:55Z

core/src/test/java/org/infinispan/xsite/AbstractTwoSitesTest.java

+   public AbstractTwoSitesTest use2Pc(boolean use2Pc) {
+      this.use2Pc = use2Pc;
+      return this;
+   }


This should be grouped with the other parameters (and with parameters() as well).

danberindei · 2017-07-06T08:51:03Z

core/src/test/java/org/infinispan/xsite/backupfailure/NonTxBackupFailureTest.java

+         // is not written. This happens when the failure is thrown on remote primary owner - we don't
+         // commit local entries until distribution interceptor returns and this now throws an exception.
+         // This used to work when we were replicating cross-site from origin only after everything was
+         // committed - the replication failure then did not affect local cluster state.


Isn't that the case now? BaseBackupInterceptor is still before EntryWrappingInterceptor...

In the past:

origin -> primary

primary commit

origin commit

origin backup -> exception, but data are committed locally

now:

origin -> primary

primary commit

primary backups -> exception

origin gets exception, does not commit

danberindei · 2017-07-06T08:54:35Z

...rc/test/java/org/infinispan/xsite/statetransfer/failures/SiteConsumerTopologyChangeTest.java

@@ -55,6 +55,7 @@ public void testXSiteDuringJoin() throws InterruptedException, ExecutionExceptio
      doXSiteStateTransferDuringTopologyChange(TopologyEvent.JOIN);
   }

+   // TODO: this test is flaky - ISPN-6872


Perhaps you should move it to the "unstable_xsite" group.

danberindei · 2017-07-06T12:29:58Z

core/src/test/java/org/infinispan/xsite/CacheOperationsTest.java

+      return Stream.of(keys).collect(Collectors.toMap(Function.identity(), ignored -> value));
+   }
+}
+


checkstyle error

pruivo

minor comments.

pruivo · 2017-07-12T13:55:24Z

core/src/main/java/org/infinispan/xsite/CustomFailurePolicy.java

 import java.util.function.Function;

 import javax.transaction.Transaction;

 import org.infinispan.Cache;
+import org.infinispan.functional.EntryView;


unused imports

pruivo · 2017-07-12T13:55:50Z

core/src/main/java/org/infinispan/marshall/core/MarshallableFunctions.java

@@ -6,10 +6,13 @@
 import java.util.function.Consumer;
 import java.util.function.Function;

+import org.infinispan.container.entries.InternalCacheValue;
+import org.infinispan.functional.EntryView;


unused import

pruivo · 2017-07-12T13:56:25Z

core/src/main/java/org/infinispan/marshall/exts/MetaParamExternalizers.java

@@ -6,10 +6,12 @@
 import java.util.Set;

 import org.infinispan.container.versioning.EntryVersion;
+import org.infinispan.functional.MetaParam;


unused import

pruivo · 2017-07-12T13:59:07Z

core/src/main/java/org/infinispan/functional/impl/MetaParams.java

+   }
+
+   public MetaParam.Writable[] toWritableMetas() {
+      int writable = 0;


suggestion: use stream instead?

return Arrays.stream(metas).filter(metaParam -> metaParam instanceof MetaParam.Writable).toArray(MetaParam.Writable[]::new);

That probably creates another intermediate array, and this is on the hot path...

Btw., I think that @karesti will introduce Metadata setter to the func API, and this will be another use case for that since the double conversion is unfortunate.

merge PR should be merged for this to happen, I guess

Yes, but none should block each other. The current solution is fine, if we decide to not include the setter.

pruivo · 2017-07-12T14:00:26Z

core/src/main/java/org/infinispan/functional/impl/MetaParamsInternalMetadata.java

@@ -6,6 +6,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.StreamSupport;


unused import

* Fix handling of ComputeCommand and ComputeIfAbsentCommand * In transaction replicate values from transactional context (replaying modifications in remote site may have different results)

rvansa · 2017-07-18T13:55:44Z

Fixed or commented.

rvansa · 2017-07-20T11:44:44Z

Accidentally merged myself :-/

rvansa added the Ready for Review label Jun 26, 2017

rvansa requested a review from pruivo June 26, 2017 12:54

pruivo added Needs Rebase and removed Ready for Review labels Jun 30, 2017

pruivo requested changes Jun 30, 2017

View reviewed changes

rvansa force-pushed the ISPN-7961 branch from 1d31010 to d69ec66 Compare July 3, 2017 19:02

rvansa added Ready for Review and removed Needs Rebase labels Jul 3, 2017

danberindei suggested changes Jul 6, 2017

View reviewed changes

karesti mentioned this pull request Jul 6, 2017

ISPN-7752 merge method #5227

Closed

tristantarrant added this to the 9.1.0.Final milestone Jul 6, 2017

danberindei reviewed Jul 6, 2017

View reviewed changes

pruivo requested changes Jul 12, 2017

View reviewed changes

pruivo added Changes Suggested and removed Ready for Review labels Jul 12, 2017

tristantarrant modified the milestones: 9.2.0.Alpha1, 9.1.0.Final Jul 12, 2017

rvansa added 2 commits July 18, 2017 14:41

Remove unused imports

8e40b23

ISPN-7961 Cross-site replication of functional commands

55e4f71

* Fix handling of ComputeCommand and ComputeIfAbsentCommand * In transaction replicate values from transactional context (replaying modifications in remote site may have different results)

rvansa force-pushed the ISPN-7961 branch from d69ec66 to 55e4f71 Compare July 18, 2017 13:54

rvansa added Ready for Review and removed Changes Suggested labels Jul 18, 2017

danberindei added Changes Suggested and removed Ready for Review labels Jul 19, 2017

rvansa merged commit 337f66f into infinispan:master Jul 20, 2017

rvansa mentioned this pull request Jul 20, 2017

Revert "ISPN-7961 Cross-site replication of functional commands" #5307

Merged

rvansa mentioned this pull request Jul 20, 2017

ISPN-7961 Cross-site replication of functional commands #5309

Merged

		@@ -122,7 +131,36 @@ protected String key(String site) {
		}


		public CacheOperationsTest cacheMode(CacheMode cacheMode) {

ISPN-7961 Cross-site replication of functional commands #5232

ISPN-7961 Cross-site replication of functional commands #5232

Conversation

rvansa commented Jun 26, 2017

pruivo commented Jun 30, 2017

pruivo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pruivo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rvansa commented Jul 18, 2017

rvansa commented Jul 20, 2017