[[execute_code_grid]]
== 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 {brandname}. However {brandname} can provide many more benefits
that aren't immediately apparent. Since {brandname} 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.

NOTE: This section covers only executing code in the grid using an embedded cache,
if you are using a remote cache you should check out link:#execute_code_remote_grid[Executing code in the Remote Grid].

=== 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
link:{javadocroot}/org/infinispan/manager/ClusterExecutor.html[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 using Java 8 and such has functional APIs in mind, thus all methods take a functional
inteface 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.

==== 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.

[source,java]
.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 link:#server_hinting[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 previous 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.

[source,java]
.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(...)
----

==== 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
link:{javadocroot}/org/infinispan/manager/ClusterExecutor.html#timeout-long-java.util.concurrent.TimeUnit-[timeout]
method and supplying the desired duration.

==== 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.

[source,java]
.SingleNode.java
----
   EmbeddedCacheManager manager = ...;
   manager.executor().singleNodeSubmission().submit(...)
----

===== 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
link:{javadocroot}/org/infinispan/manager/ClusterExecutor.html#singleNodeSubmission-int-[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.

==== 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 {brandname} cluster. Note this will
work in a cluster of 1 was well, but will be slower.


[source,java]
----
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);
   }
}
----


include::streams.adoc[]
include::locked_streams.adoc[]
include::distributed_executor.adoc[]
