
[[user-building-blocks]]
== Building Blocks

Building blocks are layered on top of channels, and can be used instead of channels whenever
a higher-level interface is required.

Whereas channels are simple socket-like constructs, building blocks may offer a far more sophisticated
interface. In some cases, building blocks offer access to the underlying channel, so that -- if the building
block at hand does not offer a certain functionality -- the channel can be accessed directly. Building blocks
are located in the `org.jgroups.blocks` package.
    

[[MessageDispatcher]]
=== MessageDispatcher

A channel is a simple class to _asynchronously_
send and receive messages. However, a significant number of communication patterns in group communication
require _synchronous_ communication. For example, a sender would like to send a message to all members of
the group and wait for all responses. Or another application would like to send a message to the group and
wait only until the majority of the receivers have sent a response, or until a timeout occurred.

MessageDispatcher provides blocking (and non-blocking) request sending and response
correlation. It offers synchronous (as well as asynchronous) message sending with request-response
correlation, e.g. matching one or multiple responses with the original request.

An example of using this class would be to send a request message to all cluster members, and block until all
responses have been received, or until a timeout has elapsed.

Contrary to <<RpcDispatcher,RpcDispatcher>>, MessageDispatcher deals with
_sending message requests and correlating message responses_, while RpcDispatcher deals
with _invoking method calls and correlating responses_. RpcDispatcher extends
MessageDispatcher, and offers an even higher level of abstraction over MessageDispatcher.

RpcDispatcher is essentially a way to invoke remote procedure calls (RCs) across a cluster.

Both MessageDispatcher and RpcDispatcher sit on top of a channel; therefore an instance of
MessageDispatcher is created with a channel as argument. It can now be
used in both __client and server role__: a client sends requests and receives responses and
a server receives requests and sends responses. MessageDispatcher allows for an
application to be both at the same time. To be able to serve requests in the server role, the
`RequestHandler.handle()` method has to be implemented:

[source,java]
----
Object handle(Message msg) throws Exception;
----

The `handle()` method is called whenever a request is received. It must return a value
(must be serializable, but can be null) or throw an exception. The returned value will be sent to the sender,
and exceptions are also propagated to the sender.

Before looking at the methods of MessageDispatcher, let's take a look at RequestOptions first.
      

[[RequestOptions]]
==== RequestOptions

Every message sending in MessageDispatcher or request invocation in RpcDispatcher is governed by an
instance of RequestOptions. This is a class which can be passed to a call to define the various
options related to the call, e.g. a timeout, whether the call should block or not, the flags (see
<<MessageFlags>>) etc.

The various options are:
              
* Response mode: this determines whether the call is blocking and - if yes - how long it should block. The modes are:
  `GET_ALL`:: Block until responses from all members (minus the suspected ones) have been received.
  `GET_NONE`:: Wait for none. This makes the call non-blocking
  `GET_FIRST`:: Block until the first response (from anyone) has been received
* Timeout: number of milliseconds we're willing to block. If the call hasn't terminated after the
  timeout elapsed, a TimeoutException will be thrown. A timeout of 0 means to wait forever. The
  timeout is ignored if the call is non-blocking (mode=`GET_NONE`)
* Anycasting: if set to true, this means we'll use unicasts to individual members rather than sending
  multicasts. For example, if we have have TCP as transport, and the cluster is {A,B,C,D,E}, and we
  send a message through MessageDispatcher where dests={C,D}, and we do _not_
  want to send the request to everyone, then we'd
  set anycasting=true. This will send the request to C and D only, as unicasts, which is better if
  we use a transport such as TCP which cannot use IP multicasting (sending 1 packet to reach all
  members).
* Response filter: A RspFilter allows for filtering of responses and user-defined termination of
  a call. For example, if we expect responses from 10 members, but can return after having
  received 3 non-null responses, a RspFilter could be used. See <<RspFilter>> for
  a discussion on response filters.
* Flags: the various flags to be passed to the message, see the section on <<MessageFlags, message flags>> for details.
* Exclusion list: here we can pass a list of members (addresses) that should be excluded. For example,
  if the view is A,B,C,D,E, and we set the exclusion list to A,C then the caller will wait for
  responses from everyone except A and C. Also, every recipient that's in the exclusion list
  will discard the message.
                  
An example of how to use RequestOptions is:
          
[source,java]
----
RpcDispatcher disp;
RequestOptions opts=new RequestOptions(ResponseMode.GET_ALL, 1000L)
                 .setFlags(Message.Flag.NO_FC, Message.Flag.OOB);
Object val=disp.callRemoteMethod(target, method_call, opts);
----

The methods to send requests are:


[source,java]
----
public <T> RspList<T>
       castMessage(Collection<Address> dests, Message msg, RequestOptions opts)
       throws Exception;

public <T> CompletableFuture<RspList<T>>
       castMessageWithFuture(Collection<Address> dests, Message msg, RequestOptions opts)
       throws Exception;

public <T> T sendMessage(Message msg, RequestOptions opts) throws Exception;

public <T> CompletableFuture<T>
       sendMessageWithFuture(Message msg, RequestOptions opts) throws Exception;
----

`castMessage()` sends a message to all members defined in `dests`. If `dests` is null, the message is sent to all
members of the current cluster.

If a message is sent synchronously (defined by `opts.mode`), then `opts.timeout`
defines the maximum amount of time (in milliseconds) to wait for the responses.

`castMessage()` returns a `RspList`, which contains a map of addresses and Rsps;
there's one `Rsp` per member listed in dests.
      
A `Rsp` instance contains the response value (or null), an exception if the target `handle()` method threw
an exception, whether the target member was suspected, or not, and so on. See the example below for
more details.

`castMessageWithFuture()` returns immediately, with a `CompletableFuture`. The future
can be used to fetch the response list (now or later), and it also allows for installation of a callback
which will be invoked when the future is done.
See <<CompleteableFuture>> for details on how to use CompletableFutures.

NOTE: The message passed to the `cast*`() and `send*()` methods needs to have a `null` destination for the `cast*()`
calls and a non-`null` destination for the `send*()` methods. +
It can be any kind of message type, e.g. `ObjectMessage` or `BytesMessage`.

`sendMessage()` sends a unicast message to a single cluster member and receives the response.

The destination of the message has to be non-null (valid address of a member). The mode argument is ignored
(it is by default set to `ResponseMode.GET_FIRST`) unless it is set to `GET_NONE` in which case
the request becomes asynchronous, ie. we will not wait for the response.
      
`sendMessageWithFuture()` returns immediately with a future, which can be used to fetch the result.

One advantage of using this building block is that failed members are removed from the set of expected
responses. For example, when sending a message to 10 members and waiting for all responses, and 2 members
crash before being able to send a response, the call will return with 8 valid responses and 2 marked as
failed. The return value of castMessage() is a RspList which contains all responses (not all methods shown):

[source,java]
----
public class RspList<T> implements Map<Address,Rsp> {
    public static boolean       isReceived(Address sender);
    public static int           numSuspectedMembers();
    public List<T>       getResults();
    public static List<Address> getSuspectedMembers();
    public static boolean       isSuspected(Address sender);
    public static Object        get(Address sender);
    public static int           size();
}
----

`isReceived()` checks whether a response from sender
has already been received. Note that this is only true as long as no response has yet been received, and the
member has not been marked as failed. `numSuspectedMembers()` returns the number of
members that failed (e.g. crashed) during the wait for responses. `getResults()`
returns a list of return values. `get()` returns the return value for a specific member.
      

[[MessageDispatcherDests]]
==== Requests and target destinations

When a non-null list of addresses is passed (as the destination list) to `MessageDispatcher.castMessage()` or
`RpcDispatcher.callRemoteMethods()`, then this does _not_ mean that only the members
included in the list will receive the message, but rather it means that we'll only wait for responses from
those members, if the call is blocking.

If we want to restrict the reception of a message to the destination members, there are a few ways to do this:
            
* If we only have a few destinations to send the message to, use several unicasts.
* Use anycasting. E.g. if we have a membership of `{A,B,C,D,E,F}`, but only want A and C to receive the
  message, then set the destination list to A and C and enable anycasting in the RequestOptions passed
  to the call (see above). This means that the transport will send 2 unicasts.
* Use exclusion lists. If we have a membership of `{A,B,C,D,E,F}`, and want to send a message to almost
  all members, but exclude D and E, then we can define an exclusion list: this is done by
  settting the destination list to `null` (= send to all members), or to `{A,B,C,D,E,F}` and set the
  exclusion list in the RequestOptions passed to the call to D and E.
                
        

[[MessageDispatcherExample]]
==== Example

This section shows an example of how to use a MessageDispatcher.

[source,java]
----
public class MessageDispatcherTest implements RequestHandler {
    JChannel          channel;
    MessageDispatcher disp;
    RspList           rsp_list;
    String            props; // to be set by application programmer

    public void start() throws Exception {
        channel=new JChannel(props);
        disp=new MessageDispatcher(channel, this);
        channel.connect("MessageDispatcherTestGroup");

        for(int i=0; i < 10; i++) {
            Util.sleep(100);
            System.out.println("Casting message #" + i);
            byte[] pl=("Number #" + i).getBytes();
            rsp_list=disp.castMessage(null,
                                      new BytesMessage(null, pl, 0, pl.length),
                                      RequestOptions.SYNC());
            System.out.println("Responses:\n" +rsp_list);
        }
        Util.close(disp,channel);
    }

    public static Object handle(Message msg) throws Exception {
        System.out.println("handle(): " + msg);
        return "Success!";
    }

    public static void main(String[] args) {
        try {
            new MessageDispatcherTest().start();
        }
        catch(Exception e) {
            System.err.println(e);
        }
    }
}
----

The example starts with the creation of a channel. Next, an instance of
MessageDispatcher is created on top of the channel. Then the channel is connected. The
MessageDispatcher will from now on send requests, receive matching responses
(client role) and receive requests and send responses (server role).
        
We then send 10 messages to the group and wait for all responses. The timeout
argument is 0, which causes the call to block until all responses have been received.
        
The `handle()` method simply prints out a message and returns a string. This will
be sent back to the caller as a response value (in `Rsp.value`). Had the call thrown an exception,
`Rsp.exception` would be set instead.
        
Finally both the MessageDispatcher and channel are closed.
        

[[RpcDispatcher]]
=== RpcDispatcher

`RpcDispatcher` extends `MessageDispatcher`. It allows a
programmer to invoke remote methods in all (or single) cluster members and optionally wait for the return
value(s). An application will typically create a channel first, and then create an
RpcDispatcher on top of it. RpcDispatcher can be used to invoke remote methods
(client role) and at the same time be called by other members (server role).

Compared to MessageDispatcher, no `handle()` method needs to be implemented. Instead the methods to be called can be
placed directly in the class using regular method definitions (see example below).
The methods will get invoked using reflection.

To invoke remote method calls (unicast and multicast) the following methods are used:


[source,java]
----

public <T> RspList<T>
       callRemoteMethods(Collection<Address> dests, String method_name, Object[] args,
                         Class[] types, RequestOptions options) throws Exception;
public <T> RspList<T>
       callRemoteMethods(Collection<Address> dests, MethodCall method_call,
                         RequestOptions opts) throws Exception;

public <T> CompletableFuture<RspList<T>>
       callRemoteMethodsWithFuture(Collection<Address> dests, MethodCall method_call,
                                   RequestOptions options) throws Exception;

public <T> T
       callRemoteMethod(Address dest, String meth, Object[] args, Class[] types,
                        RequestOptions opts) throws Exception;

public <T> T
       callRemoteMethod(Address dest,
                        MethodCall call,
                        RequestOptions options) throws Exception;

public <T> CompletableFuture<T>
       callRemoteMethodWithFuture(Address dest,
                                  MethodCall call,
                                  RequestOptions opts) throws Exception
----

The family of `callRemoteMethods()` methods is invoked with a list of receiver
addresses. If null, the method will be invoked in all cluster members (including the sender). Each call takes
the target members to invoke it on (`null` mean invoke on all cluster members), a method and a `RequestOptions` instance.

The method can be given as (1) the method name, (2) the arguments and (3) the argument types, or a
MethodCall (containing a `java.lang.reflect.Method` and argument) can be given instead.

As with MessageDispatcher, a `RspList` or a future to a RspList is returned.

The family of `callRemoteMethod()` methods takes almost the same parameters, except that there is only one destination
address instead of a list. If the dest argument is null, the call will fail.

The `callRemoteMethod()` calls return the actual result (of type T), or throw an
exception if the method threw an exception on the target member.

Java's Reflection API is used to find the correct method in the target member according to the method name and
number and types of supplied arguments. There is a runtime exception if a method cannot be resolved.


==== MethodLookup and MethodDispatcher
Using reflection to find and invoke methods is rather slow.

As an alternative, we can use method IDs and the `MethodLookup` or `MethodInvoker` interfaces to resolve
methods, which is faster and has every RPC carry less data across the wire.

Interface `MethodLookup` looks as follows:

[source,java]
----
public interface MethodLookup {
    Method findMethod(short id);
}
----

An implementation is given an ID and needs to return the associated `Method` object. Implementations typically maintain
the ID-method mappings in a hashmap and use the method ID as key into the map. This hashmap lookup is faster than
having to use Java reflection to find the method for every invocation.

A example of how to use a `MethodLookup` implementation is shown in
https://github.com/belaban/JGroups/blob/master/tests/other/org/jgroups/tests/RpcDispatcherSpeedTest.java[RpcDispatcherSpeedTest].

A `MethodLookup` still uses reflection to invoke the method against the target object. In some cases
(e.g. http://quarkus.io[Quarkus]), reflection is forbidden, or at least all methods to be invoked via reflection have
to be listed at compile-time (when generating the native image). This is tedious when adding/removing fields/methods,
and so a way of invoking methods completely free of Java reflection has been added to RpcDispatcher:

NOTE: `MethodInvoker` was added in 4.1.0

[source,java]
----
public interface MethodInvoker {
    /**
     * Invokes a method associated with a given ID and the given args against the target
     * @param target The object against which to invoke the method
     * @param id The ID of the method
     * @param args The arguments of the invocation
     * @return The result. It may be null if a method returns void
     * @throws Exception Thrown if the invocation threw an exception
     */
    Object invoke(Object target, short id, Object[] args) throws Exception;
}
----

An implementation can be set in the `RpcDispatcher` using `setMethodInvoker(MethodInvoker mi)`. When a `MethodInvoker`
is present in an `RpcDispatcher`, it takes precedence over `MethodLookup`.

A `MethodInvoker` is given the target object, against which to invoke the method, the ID of the method to invoke and
a list of arguments. A typical implementation might do the following (copied from
https://github.com/belaban/JGroups/blob/master/tests/perf/org/jgroups/tests/perf/ProgrammaticUPerf.java[ProgrammaticUPerf]):

[source,java]
----
 public Object invoke(Object target, short id, Object[] args) throws Exception {
        ProgrammaticUPerf uperf=(ProgrammaticUPerf)target;
        Boolean bool_val;
        switch(id) {
            case START:
                return uperf.startTest();
            case GET:
                Integer key=(Integer)args[0];
                return uperf.get(key);
            case PUT:
                key=(Integer)args[0];
                byte[] val=(byte[])args[1];
                uperf.put(key, val);
                return null;
            case GET_CONFIG:
                return uperf.getConfig();
            case SET_SYNC:
                uperf.setSync((Boolean)args[0]);
                return null;
            case SET_OOB:
                bool_val=(Boolean)args[0];
                uperf.setOOB(bool_val);
                return null;
            ...
            case QUIT_ALL:
                uperf.quitAll();
                return null;
        }
    }
----

The downside here is that this code needs to be changed when methods are added or removed, or when signatures change.
However, if Java reflection cannot be used, then this may be feasible.


      

[[RpcDispatcherExample]]
==== Example of using RpcDispatcher

The code below shows an example of using RpcDispatcher:


[source,java]
----

public class RpcDispatcherTest {
    JChannel           channel;
    RpcDispatcher disp;
    RspList            rsp_list;
    String             props; // set by application

    public static int print(int number) throws Exception {
        return number * 2;
    }

    public void start() throws Exception {
        MethodCall call=new MethodCall(getClass().getMethod("print", int.class));
        RequestOptions opts=new RequestOptions(ResponseMode.GET_ALL, 5000);
        channel=new JChannel(props);
        disp=new RpcDispatcher(channel, this);
        channel.connect("RpcDispatcherTestGroup");

        for(int i=0; i < 10; i++) {
            Util.sleep(100);
            call.setArgs(i);
            rsp_list=disp.callRemoteMethods(null, call, opts);
            System.out.println("Responses: " + rsp_list);
        }
        Util.close(disp, channel);
    }

    public static void main(String[] args) throws Exception {
        new RpcDispatcherTest().start();
    }
}
----

Class RpcDispatcher defines method `print()` which will be called subsequently. The entry point `start()` creates a
channel and an RpcDispatcher which is layered on top. Method `callRemoteMethods()` then invokes the remote `print()`
in all cluster members (also in the caller). When all responses have been received, the call returns
and the responses are printed.

As can be seen, the RpcDispatcher building block reduces the amount of code that
needs to be written to implement RPC-based group communication applications by providing a higher
abstraction level between the application and the primitive channels.
        

[[CompleteableFuture]]
===== Asynchronous calls with futures

When invoking a synchronous call, the calling thread is blocked until the response (or responses) has been received.

A _Future_ allows a caller to return immediately and grab the result(s) later. The methods which return futures are:


[source,java]
----
public <T> CompletableFuture<RspList<T>>
       callRemoteMethodsWithFuture(Collection<Address> dests,
                                   MethodCall method_call,
                                   RequestOptions options) throws Exceptio;
public <T> CompleteableFuture<T>
       callRemoteMethodWithFuture(Address dest,
                                  MethodCall call,
                                  RequestOptions options) throws Exception;
----

A `CompleteableFuture` extends `java.util.concurrent.Future`, with its regular methods such as `isDone()`,
`get()` and `cancel()`. CompleteableFuture also allows to install some code that is run when the future is done.
This is shown in the following code:
            
[source,java]
----
CompleteableFuture<RspList<Integer>> future=dispatcher.callRemoteMethodsWithFuture(...);
future.whenComplete((result,ex) -> {
    System.out.printf("result=%d\n", result);
});
----

Here, the result (an int) is printed to stdout when available. Note that we could also have received an exception
instead of a result, in which case argument `ex` would have carried the exception.


[[RspFilter]]
==== Response filters

Response filters allow application code to hook into the reception of responses from cluster members and
can let the request-response execution and correlation code know (1) wether a response is acceptable and
(2) whether more responses are needed, or whether the call (if blocking) can return. The
`RspFilter` interface looks as follows:
          
[source,java]
----

public interface RspFilter {
    boolean isAcceptable(Object response, Address sender);
    boolean needMoreResponses();
}
          
----

`isAcceptable()` is given a response value and the address of the member which sent
the response, and needs to decide whether the response is valid (should return true) or not
(should return false).
          
`needMoreResponses()` determine whether a call returns or not.

The sample code below shows how to use a RspFilter:


[source,java]
----

public void testResponseFilter() throws Exception {
    final long timeout = 10 * 1000 ;

    RequestOptions opts;
    opts=new RequestOptions(ResponseMode.GET_ALL,
                            timeout, false,
                            new RspFilter() {
                                int num=0;
                                public boolean isAcceptable(Object response,
                                                            Address sender) {
                                    boolean retval=((Integer)response).intValue() > 1;
                                    if(retval)
                                        num++;
                                    return retval;
                                }
                                public boolean needMoreResponses() {
                                    return num < 2;
                                }
                            });

    RspList rsps=disp1.callRemoteMethods(null, "foo", null, null, opts);
    System.out.println("responses are:\n" + rsps);
    assert rsps.size() == 3;
    assert rsps.numReceived() == 2;
}
          
----

Here, we invoke a cluster wide RPC (dests=null), which blocks (mode=`GET_ALL`) for 10 seconds max
(timeout=10000), but also passes an instance of RspFilter to the call (in options).
          
The filter accepts all responses whose value is greater than 1, and returns as soon as it has received
2 responses which satisfy the above condition.
          

WARNING: If we have a RspFilter which doesn't terminate the call even if responses from all members have
         been received, we might block forever (if no timeout was given)! For example, if we have 10 members,
         and every member returns 1 or 2 as return value of foo() in the above code, then
         isAcceptable() would always return false, therefore never incrementing `num`,
         and `needMoreResponses()` would always return true; this would never terminate
         the call if it wasn't for the timeout of 10 seconds! +
         This was fixed in 3.1; a blocking call will always return if we've received as many responses as
         we have members in `dests`, regardless of what the RspFilter says.





[[AsyncInvocation]]
=== Asynchronous invocation in MessageDispatcher and RpcDispatcher

By default, a message received by a MessageDispatcher or RpcDispatcher is dispatched into application code
by calling method handle() (1) of the RequestHandler interface:

[source,java]
----
public interface RequestHandler {
    Object handle(Message msg) throws Exception;   // <1>
    default void handle(Message request, Response response) throws Exception {
        throw new UnsupportedOperationException(); // <2>
    }
}
----

In the case of RpcDispatcher, the `handle()` method (1) converts the message's contents into a method call,
invokes the method against the target object and returns the result (or throws an exception). The return value
of `handle()` is then sent back to the sender of the message.
        
The invocation is _synchronous_, ie. done on the thread responsible for dispatching this
particular message from the network up the stack all the way into the application. The thread is therefore
_unusable_ for the duration of the method invocation.
        
If the invocation takes a while, e.g. because locks are acquired or the application waits on some I/O, as
the current thread is busy, another thread will be used for a different request message. This can quickly
lead to the thread pool being exhausted or many messages getting queued if the pool has an associated queue.
        
Therefore a new way of dispatching messages to the application was devised; the asynchronous invocation API. Method
`handle(Request,Response`) (2) takes a request message and a `Response` object.The request message contains the same
information as before (e.g. a method call plus args). The `Response` argument is used to send a reply (if needed) at
a later time, when processing is done.
        
[source,java]
----

public interface Response {
    void send(Object reply, boolean is_exception);
    void send(Message reply, boolean is_exception);
}
----

`Response` encapsulates information about the request (e.g. request ID and sender), and has method `reply()` to
send a response. The `is_exception` parameter can be set to true if the reply is actually an exception, e.g.
that was thrown when `handle()` ran application code.

The second method takes a Message which needs to carry the serialized reply in its payload. This method can be used
to control the type of message that's sent out, ie. by setting flags, adding headers and so on.

The advantage of the new API is that it can, but doesn't have to, be used asynchronously. The default
implementation still uses the synchronous invocation style:

[source,java]
----
public void handle(Message request, Response response) throws Exception {
    Object retval=handle(request);
    if(response != null)
        response.send(retval, false);
}
----

Method `handle()` is called, which synchronously calls into application code and returns a result, which is
subsequently sent back to the sender of the request message.

However, an application could subclass MessageDispatcher or RpcDispatcher (as done in Infinispan), or it
could set a custom request handler via `MessageDispatcher.setRequestHandler()`, and implement `handle()` by
dispatching the processing to a thread from a thread pool. The thread which guided the request message from
the network up to this point would be therefore immediately released and could be used to process other messages.

The response would be sent whenever the invocation of application code is done, and thus the thread from
the thread pool would not be blocked on I/O, trying to acquire locks or anything else that blocks in
application code.
        
To set the mode which is used, method `MessageDispatcher.asyncDispatching(boolean)` can be used. This can be
changed even at runtime, to switch between sync and async invocation style.

Asynchrounous invocation is typically used in conjunction with an application thread pool. The application
knows (JGroups doesn't) which requests can be processed in parallel and which ones can't. For example,
all OOB calls could be dispatched directly to the thread pool, as ordering of OOB requests is not important,
but regular requests should be added to a queue where they are processed sequentually.
        
The main benefit here is that request dispatching (and ordering) is now under application control
_if the application wants to do that_. If not, we can still use synchronous invocation.

A good example where asynchronous invocation makes sense are replicated web sessions. If a cluster node A
has 1000 web sessions, then replication of updates across the cluster generates messages from A. Because
JGroups delivers messages from the _same_ sender _sequentially_, even
updates to unrelated web sessions are delivered in strict order.

With asynchronous invocation, the application could devise a dispatching strategy which assigns updates to
different (unrelated) web sessions to any available thread from the pool, but queues updates to the same
session, and processes those by the same thread, to provide ordering of updates to the same session. This
would speed up overall processing, as updates to a web session 1 on A don't have to wait until all
updates to an unrelated web session 2 on A have been processed.

NOTE: The asynchronous invocation API was added in JGroups 3.3.
        

[[ReplicatedHashMap]]
=== ReplicatedHashMap

This class was written as a demo of how state can be shared between nodes of a cluster. It has never been
heavily tested and is therefore not meant to be used in production.

A `ReplicatedHashMap` uses a concurrent hashmap internally and allows to create several
instances of hashmaps in different processes. All of these instances have exactly the same state at all
times. When creating such an instance, a cluster name determines which cluster of replicated hashmaps will
be joined. The new instance will then query the state from existing members and update itself before
starting to service requests. If there are no existing members, it will simply start with an empty state.

Modifications such as `put()`, `clear()` or
`remove()` will be propagated in orderly fashion to all replicas. Read-only requests
such as `get()` will only be invoked on the local hashmap.
        
Since both keys and values of a hashtable will be sent across the network, they have to be
serializable. Putting a non-serializable value in the map will result in an exception at marshalling time.

A `ReplicatedHashMap` allows to register for notifications, e.g. when data is
added removed. All listeners will get notified when such an event occurs. Notification is always local;
for example in the case of removing an element, first the element is removed in all replicas, which then
notify their listener(s) of the removal (after the fact).
        
`ReplicatedHashMap` allow members in a group to share common state across process and machine boundaries.
        

[[ReplCache]]
=== ReplCache

`ReplCache` is a distributed cache which - contrary to ReplicatedHashMap - doesn't replicate its values to
all cluster members, but just to selected backups.
        
A `put(K,V,R)` method has a _replication count R_ which determines
on how many cluster members key K and value V should be stored. When we have 10 cluster members, and R=3,
then K and V will be stored on 3 members. If one of those members goes down, or leaves the cluster, then a
different member will be told to store K and V. ReplCache tries to always have R cluster members store K
and V.
        
A replication count of `-1` means that a given key and value should be stored on _all_ cluster members.
        
The mapping between a key K and the cluster member(s) on which K will be stored is always deterministic, and
is computed using a _consistent hash function_.

Note that this class was written as a demo of how state can be shared between nodes of a cluster. It has
never been heavily tested and is therefore not meant to be used in production.
        

[[LockService]]
=== Cluster wide locking

`LockService` can be used to acquire locks on a cluster-wide basis; ie. only one node can acquire a given lock. E.g. if
member B acquires lock L, and member C also tries to acquire L, then C will block until B releases L
(or leaves /crashes)

The new service is implemented as a building block (`org.jgroups.blocks.locking.LockService`) and a protocol
(`CENTRAL_LOCK` or `CENTRAL_LOCK2`). `LockService` looks up the protocol and talks to it via events. If no locking
protocol is found, `LockService` won't start and will throw an exception.

The main abstraction of a distributed lock is an implementation of `java.util.concurrent.locks.Lock`.

Below is an example of how LockService is typically used:

[source,java]
----
// locking.xml needs to contain a locking protocol, e.g. CENTRAL_LOCK
JChannel ch=new JChannel("/home/bela/locking.xml");
LockService lock_service=new LockService(ch);
ch.connect("lock-cluster");
Lock lock=lock_service.getLock("mylock"); // gets a cluster-wide lock
lock.lock();
try {
    // do something with the locked resource
}
finally {
    lock.unlock();
}
----

In the example, we create a channel, then a `LockService`, then connect the channel. If the channel's
configuration doesn't include a locking protocol, an exception will be thrown.
Then we grab a lock named `"mylock"`, which we lock and subsequently unlock. If another member P had already
acquired `"mylock"`, we'd block until P released the lock, or P left the cluster or crashed.
        
Note that the owner of a lock is always a given thread in a cluster, so the owner is the JGroups address and
the thread ID. *This means that different threads inside the same JVM trying to access the same named lock
will compete for it.* If `thread-22` grabs the lock first, then `thread-5` will block until `thread-22`
releases the lock.

NOTE: If we want the lock owner to only be the address (and not the thread-id), then property
`use_thread_id_for_lock_owner` can be set to `false`. This means that all threads in a given node can lock or unlock
a given lock. Example: thread T1 locks "lock", but thread T2 can unlock it. This is _not_ the same semantics as
`java.util.concurrent.locks.Lock`, but nevertheless useful in some scenarios. (Introduced in 3.6)

JGroups includes a demo (`org.jgroups.demos.LockServiceDemo`), which can be used to interactively experiment
with distributed locks. `LockServiceDemo -h` dumps all command line options.
        
There are two protocols which provides locking: <<CENTRAL_LOCK>> and <<CENTRAL_LOCK2>>.

Note that the locking protocol has to be placed at or towards the top of the stack (close to the channel), because it
requires reliable unicasts and multicasts (e.g. provided by `UNICAST3` and `NAKACK2`).
        

[[LockingAndMerges]]
==== Locking and merges

The following scenario is susceptible to network partitioning and subsequent merging: we have a cluster
view of `{A,B,C,D}` and then the cluster splits into `{A,B}` and `{C,D}`. Assume that B and D now acquire a
lock `"mylock"`. This is what happens (with the locking protocol being `CENTRAL_LOCK`):
                
* There are 2 coordinators: A for `{A,B}` and C for `{C,D}`
* B successfully acquires `"mylock"` from A
* D successfully acquires `"mylock"` from C
* The partitions merge back into `{A,B,C,D}`. Now, only A is the coordinator, but C ceases
to be a coordinator
* Problem: D still holds a lock which should actually be invalid!
There is no easy way (via the Lock API) to 'remove' the lock from D. We could for example simply release
D's lock on `"mylock"`, but then there's no way telling D that the lock it holds is actually stale!
            
Therefore the recommended solution here is for nodes to listen to `MergeView` changes if they expect
merging to occur, and re-acquire all of their locks after a merge, e.g.:
            
[source,java]
----

Lock l1, l2, l3;
LockService lock_service;
...
public void viewAccepted(View view) {
    if(view instanceof MergeView) {
        new Thread() {
            public void run() {
                lock_service.unlockAll();
                // stop all access to resources protected by l1, l2 or l3
                // every thread needs to re-acquire the locks it holds
            }
        }.start();
    }
}
----

==== Locking and merges (updated)
With <<CENTRAL_LOCK2>>, merging of partitions is handled differently. Contrary to CENTRAL_LOCK, which has the coordinator
back up its lock tables to one or more backup members, CENTRAL_LOCK2 doesn't do this.

Instead, when the current coordinator leaves or crashes, the new coordinator fetches information about locks and pending
lock/unlock requests from all members, and then builds its lock table based on this information.

In the above scenario with both B and D holding `mylock`, in case of a merge (say A becomes the new coordinator), D
will be told that its lock `mylock` has been *revoked*. This means that D needs to force-unlock D. This can be done
in the `lockRevoked()` callback, e.g.:

[source,java]
----
LockService lock_service;
...
public void lockRevoked(String lock_name, Owner current_owner) {
    lock_service.unlockForce(lock_name);
}
----

This is maginally better than CENTRAL_LOCK, but admittedly less than ideal. Given the following code:

[source,java]
----
Lock lock=lock_service.get("mylock";
lock.lock();
try {
   // do something while the lock is held
   longRunningAction();
}
finally {
    lock.unlock
}
----

When `mylock` is revoked, `longRunningAction()` should be stopped immediately, or - even better - its changes should be
undone (like in a transaction). However, this isn't feasible and would unnecessarily complicate the code.

Here, we see that the `Lock` abstraction, as easy as it is and as often it is used *locally* (inside the same JVM),
may not be the best abstraction for a distributed setting!



[[CounterService]]
=== Cluster wide atomic counters

Cluster wide counters provide named counters (similar to AtomicLong) which can be changed atomically. Two
nodes incrementing the same counter with initial value 10 will see 11 and 12 as results, respectively.
        
To create a named counter, the following steps have to be taken:

- [x] Add protocol `COUNTER` to the top of the stack configuration
- [x] Create an instance of CounterService
- [x] Create a new or get an existing named counter
- [x] Use the counter to increment, decrement, get, set, compare-and-set etc the counter


In the first step, we add `COUNTER` to the top of the protocol stack configuration:
        
[source,java]
----

<config>
    ...
    <MFC max_credits="2M"
         min_threshold="0.4" />
    <FRAG2 frag_size="60K" />
    <COUNTER bypass_bundling="true" timeout="5000" />
</config>
        
----

Configuration of the `COUNTER` protocol is described in <<COUNTER>>.

Next, we create a `CounterService`, which is used to create and delete named counters:

[source,java]
----
ch = new JChannel(props);
CounterService counter_service = new CounterService(ch);
ch.connect("counter-cluster");
Counter counter = counter_service.getOrCreateCounter("mycounter", 1);
----

In the sample code above, we create a channel first, then create the `CounterService` referencing the channel.
Then we connect the channel and finally create a new named counter "mycounter", with an initial value of 1.
If the counter already exists, the existing counter will be returned and the initial value will be ignored.
        
CounterService doesn't consume any messages from the channel over which it is created; instead it grabs
a reference to the COUNTER protocols and invokes methods on it directly. This has the advantage that
CounterService is non-intrusive: many instances can be created over the same channel. CounterService even
co-exists with other services which use the same mechanism, e.g. LockService or ExecutionService (see above).
        
The returned counter instance implements interface Counter:
        
[source,java]
----

package org.jgroups.blocks.atomic;

public interface Counter {

    public String getName();

    /**
     * Gets the current value of the counter
     * @return The current value
     */
    public long get();

    /**
     * Sets the counter to a new value
     * @param new_value The new value
     */
    public void set(long new_value);

    /**
     * Atomically updates the counter using a CAS operation
     *
     * @param expect The expected value of the counter
     * @param update The new value of the counter
     * @return True if the counter could be updated, false otherwise
     */
    public boolean compareAndSet(long expect, long update);

    /**
     * Atomically increments the counter and returns the new value
     * @return The new value
     */
    public long incrementAndGet();

    /**
     * Atomically decrements the counter and returns the new value
     * @return The new value
     */
    public long decrementAndGet();


    /**
     * Atomically adds the given value to the current value.
     *
     * @param delta the value to add
     * @return the updated value
     */
    public long addAndGet(long delta);
}
        
----

[[CounterServiceDesign]]
==== Design

The design of COUNTER is described in detail in
https://github.com/belaban/JGroups/blob/master/doc/design/CounterService.txt[CounterService.txt].
            
In a nutshell, in a cluster the current coordinator maintains a hashmap of named counters. Members send
requests (increment, decrement etc) to it, and the coordinator atomically applies the requests and
sends back responses.

The advantage of this centralized approach is that - regardless of the size of a cluster - every
request has a constant execution cost, namely a network round trip.

A crash or leaving of the coordinator is handled as follows. The coordinator maintains a version for
every counter value. Whenever the counter value is changed, the version is incremented. For every
request that modifies a counter, both the counter value and the version are returned to the requester.
The requester caches all counter values and associated versions in its own local cache.

When the coordinator leaves or crashes, the next-in-line member becomes the new coordinator. It then
starts a reconciliation phase, and discards all requests until the reconciliation phase has completed.
The reconciliation phase solicits all members for their cached values and versions. To reduce traffic,
the request also carries all version numbers with it.

The clients return values whose versions are higher than the ones shipped by the new coordinator. The new
coordinator waits for responses from all members or timeout milliseconds. Then it updates its own
hashmap with values whose versions are higher than its own. Finally, it stops discarding requests and
sends a resend message to all clients in order to resend any requests that might be pending.

There's another edge case that also needs to be covered: if a client P updates a counter, and both P and
the coordinator crash, then the update is lost. To reduce the chances of this happening, COUNTER
can be enabled to replicate all counter changes to one or more backup coordinators. The num_backups
property defines the number of such backups. Whenever a counter was changed in the current coordinator,
it also updates the backups (asynchronously). 0 disables this.
            

