Chapter 7. Executing code in the grid The main benefit of a cache is the ability to very quickly lookup a value by its key, even across machines. In fact this use alone is probably the reason many users use Data Grid. However Data Grid can provide many more benefits that aren't immediately apparent. Since Data Grid is usually used in a cluster of machines we also have features available that can help utilize the entire cluster for performing the user's desired workload. 7.1. Cluster Executor Since you have a group of machines, it makes sense to leverage their combined computing power for executing code on all of them them. The Cache Manager comes with a nice utility that allows you to execute arbitrary code in the cluster. Note this feature requires no Cache to be used. This Cluster Executor can be retrieved by calling executor() on the EmbeddedCacheManager . This executor is retrievable in both clustered and non clustered configurations. Note The ClusterExecutor is specifically designed for executing code where the code is not reliant upon the data in a cache and is used instead as a way to help users to execute code easily in the cluster. This manager was built specifically around Java's streaming API, thus all methods take a functional interface as an argument. Also since these arguments will be sent to other nodes they need to be serializable. We even used a nice trick to ensure our lambdas are immediately Serializable. That is by having the arguments implement both Serializable and the real argument type (ie. Runnable or Function). The JRE will pick the most specific class when determining which method to invoke, so in that case your lambdas will always be serializable. It is also possible to use an Externalizer to possibly reduce message size further. The manager by default will submit a given command to all nodes in the cluster including the node where it was submitted from. You can control on which nodes the task is executed on by using the filterTargets methods as is explained in the section. 7.1.1. Filtering execution nodes It is possible to limit on which nodes the command will be ran. For example you may want to only run a computation on machines in the same rack. Or you may want to perform an operation once in the local site and again on a different site. A cluster executor can limit what nodes it sends requests to at the scope of same or different machine, rack or site level. SameRack.java EmbeddedCacheManager manager = ...; manager.executor().filterTargets(ClusterExecutionPolicy.SAME_RACK).submit(...) To use this topology base filtering you must enable topology aware consistent hashing through Server Hinting. You can also filter using a predicate based on the Address of the node. This can also be optionally combined with topology based filtering in the code snippet. We also allow the target node to be chosen by any means using a Predicate that will filter out which nodes can be considered for execution. Note this can also be combined with Topology filtering at the same time to allow even more fine control of where you code is executed within the cluster. Predicate.java EmbeddedCacheManager manager = ...; // Just filter manager.executor().filterTargets(a -> a.equals(..)).submit(...) // Filter only those in the desired topology manager.executor().filterTargets(ClusterExecutionPolicy.SAME_SITE, a -> a.equals(..)).submit(...) 7.1.2. Timeout Cluster Executor allows for a timeout to be set per invocation. This defaults to the distributed sync timeout as configured on the Transport Configuration. This timeout works in both a clustered and non clustered Cache Manager. The executor may or may not interrupt the threads executing a task when the timeout expires. However when the timeout occurs any Consumer or Future will be completed passing back a TimeoutException . This value can be overridden by ivoking the timeout method and supplying the desired duration. 7.1.3. Single Node Submission Cluster Executor can also run in single node submission mode instead of submitting the command to all nodes it will instead pick one of the nodes that would have normally received the command and instead submit it it to only one. Each submission will possibly use a different node to execute the task on. This can be very useful to use the ClusterExecutor as a java.util.concurrent.Executor which you may have noticed that ClusterExecutor implements. SingleNode.java EmbeddedCacheManager manager = ...; manager.executor().singleNodeSubmission().submit(...) 7.1.3.1. Failover When running in single node submission it may be desirable to also allow the Cluster Executor handle cases where an exception occurred during the processing of a given command by retrying the command again. When this occurs the Cluster Executor will choose a single node again to resubmit the command to up to the desired number of failover attempts. Note the chosen node could be any node that passes the topology or predicate check. Failover is enabled by invoking the overridden singleNodeSubmission method. The given command will be resubmitted again to a single node until either the command completes without exception or the total submission amount is equal to the provided failover count. 7.1.4. Example: PI Approximation This example shows how you can use the ClusterExecutor to estimate the value of PI. Pi approximation can greatly benefit from parallel distributed execution via Cluster Executor. Recall that area of the square is Sa = 4r2 and area of the circle is Ca=pi*r2. Substituting r2 from the second equation into the first one it turns out that pi = 4 * Ca/Sa. Now, image that we can shoot very large number of darts into a square; if we take ratio of darts that land inside a circle over a total number of darts shot we will approximate Ca/Sa value. Since we know that pi = 4 * Ca/Sa we can easily derive approximate value of pi. The more darts we shoot the better approximation we get. In the example below we shoot 1 billion darts but instead of "shooting" them serially we parallelize work of dart shooting across the entire Data Grid cluster. Note this will work in a cluster of 1 was well, but will be slower. public class PiAppx { public static void main (String [] arg){ EmbeddedCacheManager cacheManager = .. boolean isCluster = .. int numPoints = 1_000_000_000; int numServers = isCluster ? cacheManager.getMembers().size() : 1; int numberPerWorker = numPoints / numServers; ClusterExecutor clusterExecutor = cacheManager.executor(); long start = System.currentTimeMillis(); // We receive results concurrently - need to handle that AtomicLong countCircle = new AtomicLong(); CompletableFuture<Void> fut = clusterExecutor.submitConsumer(m -> { int insideCircleCount = 0; for (int i = 0; i < numberPerWorker; i++) { double x = Math.random(); double y = Math.random(); if (insideCircle(x, y)) insideCircleCount++; } return insideCircleCount; }, (address, count, throwable) -> { if (throwable != null) { throwable.printStackTrace(); System.out.println("Address: " + address + " encountered an error: " + throwable); } else { countCircle.getAndAdd(count); } }); fut.whenComplete((v, t) -> { // This is invoked after all nodes have responded with a value or exception if (t != null) { t.printStackTrace(); System.out.println("Exception encountered while waiting:" + t); } else { double appxPi = 4.0 * countCircle.get() / numPoints; System.out.println("Distributed PI appx is " + appxPi + " using " + numServers + " node(s), completed in " + (System.currentTimeMillis() - start) + " ms"); } }); // May have to sleep here to keep alive if no user threads left } private static boolean insideCircle(double x, double y) { return (Math.pow(x - 0.5, 2) + Math.pow(y - 0.5, 2)) <= Math.pow(0.5, 2); } }

Commands:
EmbeddedCacheManager manager = ...; manager.executor().filterTargets(ClusterExecutionPolicy.SAME_RACK).submit(...)
EmbeddedCacheManager manager = ...; // Just filter manager.executor().filterTargets(a -> a.equals(..)).submit(...) // Filter only those in the desired topology manager.executor().filterTargets(ClusterExecutionPolicy.SAME_SITE, a -> a.equals(..)).submit(...)
EmbeddedCacheManager manager = ...; manager.executor().singleNodeSubmission().submit(...)
public class PiAppx { public static void main (String [] arg){ EmbeddedCacheManager cacheManager = .. boolean isCluster = .. int numPoints = 1_000_000_000; int numServers = isCluster ? cacheManager.getMembers().size() : 1; int numberPerWorker = numPoints / numServers; ClusterExecutor clusterExecutor = cacheManager.executor(); long start = System.currentTimeMillis(); // We receive results concurrently - need to handle that AtomicLong countCircle = new AtomicLong(); CompletableFuture<Void> fut = clusterExecutor.submitConsumer(m -> { int insideCircleCount = 0; for (int i = 0; i < numberPerWorker; i++) { double x = Math.random(); double y = Math.random(); if (insideCircle(x, y)) insideCircleCount++; } return insideCircleCount; }, (address, count, throwable) -> { if (throwable != null) { throwable.printStackTrace(); System.out.println("Address: " + address + " encountered an error: " + throwable); } else { countCircle.getAndAdd(count); } }); fut.whenComplete((v, t) -> { // This is invoked after all nodes have responded with a value or exception if (t != null) { t.printStackTrace(); System.out.println("Exception encountered while waiting:" + t); } else { double appxPi = 4.0 * countCircle.get() / numPoints; System.out.println("Distributed PI appx is " + appxPi + " using " + numServers + " node(s), completed in " + (System.currentTimeMillis() - start) + " ms"); } }); // May have to sleep here to keep alive if no user threads left } private static boolean insideCircle(double x, double y) { return (Math.pow(x - 0.5, 2) + Math.pow(y - 0.5, 2)) <= Math.pow(0.5, 2); } }