By default, each Coherence cluster with the coherence-concurrent module on the classpath includes a single-threaded executor that may be used to run the dispatched tasks.

RemoteExecutor remoteExecutor = RemoteExecutor.getDefault();

Future<Void> result = remoteExecutor.submit(() -> System.out.println("Executed"));

result.get(); // block until completion

RemoteExecutor remoteExecutor = RemoteExecutor.get("Fixed5");

RejectedExecutionException

Each RemoteExecutor instance may hold local resources that should be released when the RemoteExecutor is no longer required. Like an ExecutorService, a RemoteExecutor has similar methods to shut down the executor. Calling these methods has no impact on the executors registered within the cluster.

While the RemoteExecutor does provide functionality similar to the standard ExecutorService included in the JDK, this may not be enough in the context of Coherence. A task might need to run across multiple Coherence members, produce intermediate results, and remain durable in case a cluster member running the task fails.

In such cases, task orchestration can be used. Before diving into the details of orchestration, the following concepts should be understood.

• Tasks
  
• Task Context
  
• Task Orchestration
  
• Task Collector and Collectable
  
• Task Coordinator
  
• Task Subscriber
  

To begin, consider the following code common to orchestration examples:

// demonstrate orchestration using the default RemoteExecutor
RemoteExecutor executor = RemoteExecutor.getDefault();

// WaitingSubscriber is an implementation of the
// com.oracle.coherence.concurrent.executor.Task.Subscriber interface
// that has a get() method that blocks until Subscriber.onComplete() is
// called and will return the results received by onNext()
WaitingSubscriber subscriber = new WaitingSubscriber();

// ValueTask is an implementation of the
// com.oracle.coherence.concurrent.executor.Task interface
// that returns the value provided at construction time
ValueTask task = new ValueTask("Hello World");

Given the previous example, the simplest example of an orchestration is:

// orchestrate the task, subscribe, and submit
executor.orchestrate(task)
        .subscribe(subscriber)
        .submit();

// wait for the task to complete
// if this was run on four cluster members running the default executor service,
// the returned Collection will have four results
Collection<String> results = subscriber.get();

Building on the previous example, assume a cluster with two storage and two proxy members. The cluster members are configured with the roles of storage and proxy, respectively. Let's say the task needs to run on storage members only, then the orchestration could look like:

// orchestrate the task, filtering by a role, subscribe, and submit
executor.orchestrate(task)
        .filter(Predicates.role("storage"))
        .subscribe(subscriber)
        .submit();

// wait for the task to complete
// as there are only two storage members in this hypothetical, only two
// results will be returned
Collection<String> results = subscriber.get();

There are several predicates available for use in com.oracle.coherence.concurrent.executor.function.Predicates, however, in case none apply to the target use case, simply implement the Remote.Predicate interface.

You can customize the collection of results and how they are presented to the subscriber by using collect(Collector) and until(Predicate):

// orchestrate the task, collecting the first non-null result,
// subscribe, and submit
executor.orchestrate(new MayReturnNullTask())
        .collect(TaskCollectors.firstOf())
        .until(Predicates.nonNullValue())
        .subscribe(subscriber)
        .submit();

// wait for the task to complete
// the first non-result returned will be the one provided to the
// subscriber
Collection<String> results = subscriber.get();

Several collectors are provided in com.oracle.coherence.concurrent.executor.TaskCollectors, however, in case none apply to the target use case, implement the Task.Collector interface.

You can inject RemoteExecutors through CDI.

@Inject
private RemoteExecutor single;    // injects a RemoteExecutor named 'single'.

@Inject
@Name("Fixed5")
private RemoteExecutor fixedPoolRemoteExecutor;    // injects a RemoteExecutor named 'Fixed5'. 

The instances of numeric atomic types, such as AtomicInteger and AtomicLong, are frequently used to represent various counters in the application where a client may need to increment the value, but does not necessarily need to know what the new value is.

When working with the local atomics, you can use the same API shown earlier (see Using Atomics) and simply ignore the return value. However, when using distributed atomics that would introduce unnecessary blocking on the client while waiting for the response from the server, which would then simply be discarded. Obviously, this will have a negative impact on both performance and throughput of the atomics.

To reduce the impact of remote calls in those situations, Coherence Concurrent also provides non-blocking, asynchronous implementations of all atomic types it supports.

// Creates a remote, distributed instance of named, non-blocking 'AsyncAtomicInteger', with an implicit initial value of 0.
AsyncAtomicInteger asyncFoo = Atomics.remoteAtomicInteger("foo").async();      

// Creates a remote, distributed instance of named, non-blocking 'AsyncAtomicLong', with an initial value of 5.
AsyncAtomicLong    asyncBar = Atomics.remoteAtomicLong("bar", 5L).async();

CompletableFuture<Integer> futureCounter1 = asyncFoo.incrementAndGet();
CompletableFuture<Long>    futureCounter5 = asyncBar.addAndGet(5L);

Both the blocking and non-blocking instances of any distributed atomic type, with the same name, are backed by the same cluster-side atomic instance state, so they can be used interchangeably.

Atomic types from Coherence Concurrent can also be injected using CDI, which eliminates the need for explicit factory method calls on the Atomics class.

// Injects a local, in-process instance of an 'AtomicInteger' named 'foo', with an implicit initial value of '0'.
@Inject
@Name("foo")
private AtomicInteger localFoo;   

// Injects a remote, distributed instance of an 'AtomicInteger' named 'foo', distinct from 
// the local instance 'foo', with an implicit initial value of '0'.
@Inject
@Remote
@Name("foo")
private AtomicInteger remoteFoo;  

// Injects a remote, distributed instance of non-blocking 'AsyncAtomicLong', with an implicit name of 'asyncBar'.
@Inject
@Remote
private AsyncAtomicLong asyncBar 

After you obtain an instance of an atomic type through a CDI injection, you can use it the same way you would use an instance obtained directly from the Atomics factory class.

A RemoteLock class provides an implementation of a Lock interface and enables you to ensure that only one thread on one member is running a critical section guarded by the lock at any given time.

Lock foo = Locks.remoteLock("foo");

Lock foo = Locks.localLock("foo");

foo.lock();
try {
    // critical section guarded by the exclusive lock `foo`
}
finally {
    foo.unlock();
}

A RemoteReadWriteLock class provides an implementation of a ReadWriteLock interface and enables you to ensure that only one thread on one member is running a critical section guarded by the write lock at any given time, while allowing multiple concurrent readers.

ReadWriteLock bar = Locks.remoteReadWriteLock("bar");

ReadWriteLock bar = Locks.localReadWriteLock("bar");

bar.writeLock().lock()
try {
    // critical section guarded by the exclusive write lock `bar`
}
finally {
    bar.writeLock().unlock();
}

Or:

bar.readLock().lock()
try {
    // critical section guarded by the shared read lock `bar`
}
finally {
    bar.readLock().unlock();
}

You can also use CDI to inject both the exclusive and read/write lock instances into objects that need them:

// Injects distributed exclusive lock named 'foo' into the 'lock' field.
@Inject
@Remote
@Name("foo")
private Lock lock;           

// Injects distributed read/write lock named 'bar' into the 'bar' field.
@Inject
@Remote
@Name("bar")
private ReadWriteLock bar; 

After you obtain an instance of lock through a CDI injection, you can use it the same way you would use an instance obtained directly from the Locks factory class.

The RemoteCoundDownLatch class provides a distributed implementation of a CountDownLatch, and enables you to ensure that the execution of the code on any cluster member that is waiting for the latch proceeds only when the latch reaches zero. Any cluster member can both wait for a latch and count down.

// Creates an instance of a 'RemoteCountDownLatch' with the initial count of '5'.
CoundDownLatch foo = Latches.remoteCountDownLatch("foo", 5);

// Creates an instance of a 'LocalCountDownLatch' with the initial count of '10'.
CoundDownLatch foo = Latches.localCountDownLatch("foo", 10);

After you have a RemoteCountDownLatch instance, you can use it as you normally would, by calling the countDown and await methods on it.

The RemoteSemaphore class provides a distributed implementation of a Semaphore, and enables any cluster member to acquire and release permits from the same semaphore instance.

// Creates an instance of a 'RemoteSemaphore' with '5' permits.
Semaphore foo = Semaphores.remoteSemaphore("foo", 5);

// Creates an instance of a 'LocalSemaphore' with '0' permits.  
Semaphore foo = Semaphores.localSemaphore("foo");

After you create a Semaphore instance, you can use it as you normally would, by calling the release and acquire methods on it.

You can also use CDI to inject both the CountDownLatch and Semaphore instances into objects that need them:

// Injects an instance of 'LocalCountDownLatch' with the initial count of '5'.
@Inject
@Name("foo")
@Count(5)
private CountDownLatch localLatchFoo;

// Injects an instance of 'RemoteCountDownLatch' with the initial count of '10'.
@Inject
@Name("foo")
@Remote
@Count(10)
private CountDownLatch remoteLatchFoo;          

// Inject an instance of 'LocalSemaphore' with '0' (zero) permits available.
@Inject
@Name("bar")
@Remote
private Semaphore localSemaphoreBar;            

// Inject an instance of 'RemoteSemaphore' with '1' permit available.
@Inject
@Name("bar")
@Remote
@Permits(1)
private Semaphore remoteSemaphoreBar;

After you obtain a latch or a semaphore instance through a CDI injection, you can use the same way as you would use an instance obtained directly from the Latches or Semaphores factory classes.

The @Name annotation is optional in both cases as long as the member name (in the examples above, the field name) can be obtained from the injection point, but is required otherwise (such as when you use a constructor injection).

The @Count annotation specifies the initial latch count, and if omitted, will default to one. The @Permits annotation specifies the number of available permits for a semaphore, and if omitted, will default to zero, which means that the first acquire call will block until another thread releases one or more permits.