Reliable Multicasting with the JGroups Toolkit

The first method takes an object as argument and serializes it into a byte buffer (the object has to be serializable or externalizable). The byte array is then returned. This method is often used to serialize objects into the byte buffer of a message. The second method returns a reconstructed object from a buffer. Both methods throw an exception if the object cannot be serialized or unserialized.

Contrary to the pull-style of channels, some building blocks (e.g. PullPushAdapter ) provide an event-like push-style message delivery model. In this case, the entity to be notified of message reception needs to provide a callback to be invoked whenever a message has been received. The MessageListener interface below provides a method to do so:

          public interface MessageListener {
              public void receive(Message msg);
              byte[] getState();
              void setState(byte[] state);
          }
          

Method receive() will be called when a message is received. The getState() and setState() methods are used to fetch and set the group state (e.g. when joining). Refer to Section 3.6.13, “Getting the group's state” for a discussion of state transfer.

JGroups release 2.3 introduced ExtendedMessageListener enabling partial state transfer (refer to Section 3.6.15, “Partial state transfer” ) while release 2.4 further expands ExtendedMessageListener with streaming state transfer callbacks:

         public interface ExtendedMessageListener extends MessageListener {
             byte[] getState(String state_id);
             void setState(String state_id, byte[] state);
                      
             /*** since JGroups 2.4 *****/
             void getState(OutputStream ostream);
             void getState(String state_id, OutputStream ostream);
             void setState(InputStream istream);
             void setState(String state_id, InputStream istream);
          }
       

The MembershipListener interface is similar to the MessageListener interface above: every time a new view, a suspicion message, or a block event is received, the corresponding method of the class implementing MembershipListener will be called.

            public interface MembershipListener {
                public void viewAccepted(View new_view);
                public void suspect(Object suspected_mbr);
                public void block();
            }
          

Oftentimes the only method containing any functionality will be viewAccepted() which notifies the receiver that a new member has joined the group or that an existing member has left or crashed. The suspect() callback is invoked by JGroups whenever a member if suspected of having crashed, but not yet excluded ^[1].

The block() method is called to notify the member that it will soon be blocked sending messages. This is done by the FLUSH protocol, for example to ensure that nobody is sending messages while a state transfer is in progress. When block() returns, any thread sending messages will be blocked, until FLUSH unblocks the thread again, e.g. after the state has been transferred successfully.

Therefore, block() can be used to send pending messages or complete some other work.

Note that block() should be brief, or else the entire FLUSH protocol is blocked.

The ExtendedMembershipListener interface extends MembershipListener:

              public interface ExtendedMembershipListener extends MembershipListener {
                  public void unblock();
              }
          

The unblock() method is called to notify the member that the FLUSH protocol has completed and the member can resume sending messages. If the member did not stop sending messages on block(), FLUSH simply blocked them and will resume, so no action is required from a member. Implementation of the unblock() callback is optional.

           public interface ChannelListener {
               void channelConnected(Channel channel);
               void channelDisconnected(Channel channel);
               void channelClosed(Channel channel);
               void channelShunned(); // deprecated in 2.8
               void channelReconnected(Address addr); // deprecated in 2.8
           }
        

A class implementing ChannelListener can use the Channel.setChannelListener() method to register with a channel to obtain information about state changes in a channel. Whenever a channel is closed, disconnected or opened a callback will be invoked.

         public interface Receiver extends MessageListener, MembershipListener {
         }
      

A Receiver can be used to receive messages and view changes in push-style; rather than having to pull these events from a channel, they will be dispatched to the receiver as soon as they have been received. This saves one thread (application thread, pulling messages from a channel, or the PullPushAdapter thread

Note that JChannel.receive() has been deprecated and will be removed in 3.0. The preferred way of receiving messages is now via a Receiver callback (push style).

      public interface ExtendedReceiver extends ExtendedMessageListener, MembershipListener {
      }
      

This is a receiver who will be able to handle partial state transfer

These classes implement Receiver and ExtendedReceiver. When implementing a callback, one can simply extend ReceiverAdapter and overwrite receive() in order to not having to implement all callbacks of the interface.

The ViewId is used to uniquely number views. It consists of the address of the view creator and a sequence number. ViewIds can be compared for equality and put in a hashtable as they implement equals() and hashCode() methods.^[2]

Whenever a group splits into subgroups, e.g. due to a network partition, and later the subgroups merge back together, a MergeView instead of a View will be received by the application. The MergeView class is a subclass of View and contains as additional instance variable the list of views that were merged. As an example if the group denoted by view V1:(p,q,r,s,t) split into subgroups V2:(p,q,r) and V2:(s,t) , the merged view might be V3:(p,q,r,s,t) . In this case the MergeView would contains a list of 2 views: V2:(p,q,r) and V2:(s,t) .

A channel can be created in two ways: an instance of a subclass of Channel is created directly using its public constructor (e.g. new JChannel() ), or a channel factory is created, which -- upon request -- creates instances of channels. We will only look at the first method of creating channel: by direct instantiation.

The public constructor of JChannel looks as follows:

          public JChannel(String props) throws ChannelException {}
      

It creates an instance of JChannel . The props argument points to an XML file containing the configuration of the protocol stack to be used. This can be a String, but there are also other constructors which take for example a DOM element or a URL (more on this later).

If the props argument is null, the default properties will be used. An exception will be thrown if the channel cannot be created. Possible causes include protocols that were specified in the property argument, but were not found, or wrong parameters to protocols.

For example, the Draw demo can be launched as follows:

          java org.javagroups.demos.Draw -props file:/home/bela/udp.xml
        

or

          java org.javagroups.demos.Draw -props http://www.jgroups.org/udp.xml
        

In the latter case, an application downloads its protocol stack specification from a server, which allows for central administration of application properties.

A sample XML configuration looks like this (edited from udp.xml):

              <config>
                  <UDP
                       mcast_addr="${jgroups.udp.mcast_addr:228.10.10.10}"
                       mcast_port="${jgroups.udp.mcast_port:45588}"
                       discard_incompatible_packets="true"
                       max_bundle_size="60000"
                       max_bundle_timeout="30"
                       ip_ttl="${jgroups.udp.ip_ttl:2}"
                       enable_bundling="true"
                       thread_pool.enabled="true"
                       thread_pool.min_threads="1"
                       thread_pool.max_threads="25"
                       thread_pool.keep_alive_time="5000"
                       thread_pool.queue_enabled="false"
                       thread_pool.queue_max_size="100"
                       thread_pool.rejection_policy="Run"
                       oob_thread_pool.enabled="true"
                       oob_thread_pool.min_threads="1"
                       oob_thread_pool.max_threads="8"
                       oob_thread_pool.keep_alive_time="5000"
                       oob_thread_pool.queue_enabled="false"
                       oob_thread_pool.queue_max_size="100"
                       oob_thread_pool.rejection_policy="Run"/>
                  <PING timeout="2000"
                          num_initial_members="3"/>
                  <MERGE2 max_interval="30000"
                          min_interval="10000"/>
                  <FD_SOCK/>
                  <FD timeout="10000" max_tries="5" />
                  <VERIFY_SUSPECT timeout="1500"  />
                  <BARRIER />
                  <pbcast.NAKACK
                                 use_mcast_xmit="false" gc_lag="0"
                                 retransmit_timeout="300,600,1200,2400,4800"
                                 discard_delivered_msgs="true"/>
                  <UNICAST timeout="300,600,1200,2400,3600"/>
                  <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
                                 max_bytes="400000"/>
                  <VIEW_SYNC avg_send_interval="60000"   />
                  <pbcast.GMS print_local_addr="true" join_timeout="3000"
                              view_bundling="true"/>
                  <FC max_credits="20000000"
                                  min_threshold="0.10"/>
                  <FRAG2 frag_size="60000"  />
                  <pbcast.STATE_TRANSFER  />
              </config>
          

A stack is wrapped by <config> and </config> elements and lists all protocols from bottom (UDP) to top (STATE_TRANSFER). Each element defines one protocol.

Each protocol is implemented as a Java class. When a protocol stack is created based on the above XML configuration, the first element ("UDP") becomes the bottom-most layer, the second one will be placed on the first, etc: the stack is created from the bottom to the top.

Each element has to be the name of a Java class that resides in the org.jgroups.stack.protocols package. Note that only the base name has to be given, not the fully specified class name (UDP instead of org.jgroups.stack.protocols.UDP). If the protocol class is not found, JGroups assumes that the name given is a fully qualified classname and will therefore try to instantiate that class. If this does not work an exception is thrown. This allows for protocol classes to reside in different packages altogether, e.g. a valid protocol name could be com.sun.eng.protocols.reliable.UCAST .

Each layer may have zero or more arguments, which are specified as a list of name/value pairs in parentheses directly after the protocol name. In the example above, UDP is configured with some options, one of them being the IP multicast address (mcast_addr) which is set to 228.10.10.10, or to the value of the system property jgroups.udp.mcast_addr, if set.

Note that all members in a group have to have the same protocol stack.

     JChannel ch=new JChannel(false);         // 1
     ProtocolStack stack=new ProtocolStack(); // 2
     ch.setProtocolStack(stack);              // 3
     stack.addProtocol(new UDP().setValue("bind_addr", InetAddress.getByName("192.168.1.5")))
             .addProtocol(new PING())
             .addProtocol(new MERGE2())
             .addProtocol(new FD_SOCK())
             .addProtocol(new FD_ALL().setValue("timeout", 12000).setValue("interval", 3000))
             .addProtocol(new VERIFY_SUSPECT())
             .addProtocol(new BARRIER())
             .addProtocol(new NAKACK())
             .addProtocol(new UNICAST2())
             .addProtocol(new STABLE())
             .addProtocol(new GMS())
             .addProtocol(new UFC())
             .addProtocol(new MFC())
             .addProtocol(new FRAG2());       // 4
     stack.init();                            // 5

     ch.setReceiver(new ReceiverAdapter() {
         public void viewAccepted(View new_view) {
             System.out.println("view: " + new_view);
         }

         public void receive(Message msg) {
             System.out.println(msg.getObject() + " [" + msg.getSrc() + "]");
         }
     });

     ch.connect("ChatCluster");

     for(;;) {
         String line=Util.readStringFromStdin(": ");
         ch.send(null, null, line);
     }
                

A number of options can be set in a channel. To do so, the following method is used:

              public void setOpt(int option, Object value);
              

Arguments are the options number and a value. The following options are currently recognized:

The equivalent method to get options is getOpt():

          public Object getOpt(int option);
      

Given an option, the current value of the option is returned.

A channel can be given a logical name which is then used instead of the channel's address. A logical name might show the function of a channel, e.g. "HostA-HTTP-Cluster", which is more legible than a UUID 3c7e52ea-4087-1859-e0a9-77a0d2f69f29.

For example, when we have 3 channels, using logical names we might see a view "{A,B,C}", which is nicer than "{56f3f99e-2fc0-8282-9eb0-866f542ae437, ee0be4af-0b45-8ed6-3f6e-92548bfa5cde, 9241a071-10ce-a931-f675-ff2e3240e1ad} !"

If no logical name is set, JGroups generates one, using the hostname and a random number, e.g. linux-3442. If this is not desired and the UUIDs should be shown, use system property -Djgroups.print_uuids=true.

The logical name can be set using:

               public void setName(String logical_name);
            

This should be done before connecting a channel. Note that the logical name stays with a channel until the channel is destroyed, whereas a UUID is created on each connection.

When JGroups starts, it prints the logical name and the associated physical address(es):

            -------------------------------------------------------------------
            GMS: address=mac-53465, cluster=DrawGroupDemo, physical address=192.168.1.3:49932
            -------------------------------------------------------------------
            ** View=[mac-53465|0] [mac-53465]
            

The logical name is mac-53465 and the physical address is 192.168.1.3:49932. The UUID is not shown here.

Since 2.12 address generation is pluggable. This means that an application can determine what kind of addresses it uses. The default address type is UUID, and since some protocols use UUID, it is recommended to provide custom classes as subclasses of UUID.

This can be used to for example pass additional data around with an address, for example information about the location of the node to which the address is assigned. Note that methods equals(), hashCode() and compare() of the UUID super class should not be changed.

To use custom addresses, the following things have to be done:

An example of a subclass is org.jgroups.util.PayloadUUID.

When a client wants to join a group, it connects to a channel giving the name of the group to be joined:

           public void connect(String clustername) throws ChannelClosed;
      

The cluster name is a string, naming the cluster to be joined. All channels that are connected to the same name form a cluster. Messages multicast on any channel in the cluster will be received by all members (including the one who sent it ^[3] ).

The method returns as soon as the group has been joined successfully. If the channel is in the closed state (see Figure 3.2, “Channel states” ), an exception will be thrown. If there are no other members, i.e. no other member has connected to a group with this name, then a new group is created and the member joined. The first member of a group becomes its coordinator . A coordinator is in charge of multicasting new views whenever the membership changes ^[4] .

Clients can also join a cluster group and fetch cluster state in one operation. The best way to conceptualize connect and fetch state connect method is to think of it as an invocation of regular connect and getstate methods executed in succession. However, there are several advantages of using connect and fetch state connect method over regular connect. First of all, underlying message exchange is heavily optimized, especially if the flush protocol is used in the stack. But more importantly, from clients perspective, connect and join operations become one atomic operation.

		      public void connect(string cluster_name, address target,
                                  string state_id, long timeout)
                          throws ChannelException;
		      

Just as in regular connect method cluster name represents a cluster to be joined. Address parameter indicates a cluster member to fetch state from. Null address parameter indicates that state should be fetched from the cluster coordinator. If state should be fetched from a particular member other than coordinator clients can provide an address of that member. State id used for partial state transfer while timeout bounds entire join and fetch operation.

Method getLocalAddress() returns the local address of the channel^[5]. In the case of JChannel , the local address is generated by the bottom-most layer of the protocol stack when the stack is connected to. That means that -- depending on the channel implementation -- the local address may or may not be available when a channel is in the unconnected state.

         public Address getLocalAddress(); // use getAddress() with 2.8.0+
      

Method getClusterName() returns the name of the cluster in which the channel is a member:

          public String getClusterName();
      

Again, the result is undefined if the channel is in the unconnected or closed state.

The following method can be used to get the current view of a channel:

         public View getView();
      

This method does not retrieve a new view (message) from the channel, but only returns the current view of the channel. The current view is updated every time a view message is received: when method receive() is called, and the return value is a view, before the view is returned, it will be installed in the channel, i.e. it will become the current view.

Calling this method on an unconnected or closed channel is implementation defined. A channel may return null, or it may return the last view it knew of.

Once the channel is connected, messages can be sent using the send() methods:

         public void send(Message msg) throws ChannelNotConnected, ChannelClosed;
         public void send(Address dst, Address src, Object obj)
                          throws ChannelNotConnected, ChannelClosed;
      

The first send() method has only one argument, which is the message to be sent. The message's destination should either be the address of the receiver (unicast) or null (multicast). When it is null, the message will be sent to all members of the group (including itself). The source address may be null; if it is, it will be set to the channel's address (so that recipients may generate a response and send it back to the sender).

The second send() method is a helper method and uses the former method internally. It requires the address of receiver and sender and an object (which has to be serializable), constructs a Message and sends it.

If the channel is not connected, or was closed, an exception will be thrown upon attempting to send a message.

Here's an example of sending a (multicast) message to all members of a group:

          Map data; // any serializable data
          try {
              channel.send(null, null, data);
          }
          catch(Exception ex) {
              // handle errors
          }
      

The null value as destination address means that the message will be sent to all members in the group. The sender's address will be filled in by the bottom-most protocol. The payload is a hashmap, which will be serialized into the message's buffer and unserialized at the receiver's end. Alternatively, any other means of generating a byte buffer and setting the message's buffer to it (e.g. using Message.setBuffer()) would also work.

Here's an example of sending a (unicast) message to the first member (coordinator) of a group:

          Map data;
          try {
              Address receiver=channel.getView().getMembers().first();
              channel.send(receiver, null, data);
          }
          catch(Exception ex) {
              // handle errors
          }
      

It creates a Message with a specific address for the receiver (the first member of the group). Again, the sender's address can be left null as it will be filled in by the bottom-most protocol.

Method receive() is used to receive messages, views, suspicions and blocks:

         public Object receive(long timeout) throws ChannelNotConnected,
                                                  ChannelClosed, Timeout;
      

A channel receives messages asynchronously from the network and stores them in a queue. When receive() is called, the next available message from the top of that queue is removed and returned. When there are no messages on the queue, the method will block. If timeout is greater than 0, it will wait the specified number of milliseconds for a message to be received, and throw a TimeoutException exception if none was received during that time. If the timeout is 0 or negative, the method will wait indefinitely for the next available message.

Depending on the channel options (see Section 3.6.2, “Setting options” ), the following types of objects may be received:

The caller has to check the type of the object returned. This can be done using the instanceof operator, as follows:

          Object obj=channel.receive(0); // wait forever
          if(obj instanceof Message)
              Message msg=(Message)obj;
          else if(obj instanceof View)
              View v=(View)obj;
          else
              ; // don't handle suspicions or blocks
      

If for example views, suspicions and blocks are disabled, then the caller is guaranteed to only receive return values of type Message . In this case, the return value can be cast to a Message directly, without using the instanceof operator.

If the channel is not connected, or was closed, a corresponding exception will be thrown.

The example below shows how to retrieve the "Hello world" string from a message:

          Message msg; // received above
          try {
              String s=(String)msg.getObject(); // error if obj not Serializable
              // alternative: s=new String(msg.getBuffer());
          }
          catch(Exception ex) {
              // handle errors, e.g. casting error above)
          }
      

The Message.getObject() method retrieves the message's byte buffer, converts it into a (serializable) object and returns the object.

Instead of pulling messages from a channel in an application thread, a Receiver can be registered with a channel. This is the preferred and recommended way of receiving messages. In 3.0, the receive() method will be removed from JChannel. All received messages, view changes and state transfer requests will invoke callbacks on the registered Receiver:

          JChannel ch=new JChannel();
          ch.setReceiver(new ExtendedReceiverAdapter() {
              public void receive(Message msg) {
                  System.out.println("received message " + msg);
              }
              public void viewAccepted(View new_view) {
                  System.out.println("received view " + new_view);
              }
          });
          ch.connect("bla");
      

The ExtendedReceiverAdapter class implements all callbacks of ExtendedReceiver with no-ops, in the example above we override receive() and viewAccepted().

The advantage of using a Receiver is that the application doesn't have to waste 1 thread for pulling messages out of a channel. In addition, the channel doesn't have to maintain an (unbounded) queue of messages/views, which can quickly get large if the receiver cannot process messages fast enough, and the sender keeps sending messages.

Instead of removing the next available message from the channel, peek() just returns a reference to the next message, but does not remove it. This is useful when one has to check the type of the next message, e.g. whether it is a regular message, or a view change. The signature of this method is not shown here, it is the same as for receive() .

A newly joined member may wish to retrieve the state of the group before starting work. This is done with getState(). This method returns the state of one member (in most cases, of the oldest member, the coordinator). It returns true or false, depending on whether a valid state could be retrieved. For example, if a member is a singleton, then calling this method would always return false ^[6] .

The actual state is returned as the return value of one of the subsequent receive() calls, in the form of a SetStateEvent object. If getState() returned true, then a valid state (non-null) will be returned, otherwise a null state will be returned. Alternatively if an application uses MembershipListener (see Section 3.2.3, “MembershipListener” ) instead of pulling messages from a channel, the getState() method will be invoked and a copy of the current state should be returned. By the same token, setting a state would be accomplished by JGroups calling the setState() method of the state fetcher.

The reason for not directly returning the state as a result of getState() is that the state has to be returned in the correct position relative to other messages. Returning it directly would violate the FIFO properties of a channel, and state transfer would not be correct.

The following code fragment shows how a group member participates in state transfers:

          channel=new JChannel();
          channel.connect("TestChannel");
          boolean rc=channel.getState(null, 5000);

          ...

          Object state, copy;
          Object ret=channel.receive(0);
          if(ret instanceof Message)
              ;
          else if(ret instanceof GetStateEvent) {
              // make a copy so that other msgs don't change the state
              copy=copyState(state);
              channel.returnState(Util.objectToByteBuffer(copy));
          }
          else if(ret instanceof SetStateEvent) {
              SetStateEvent e=(SetStateEvent)ret;
              state=e.getArg();
          }
      

A JChannel has to be created whose stack includes the STATE_TRANSFER or pbcast.STATE_TRANSFER protocols (see Chapter 5, Advanced Concepts ). Method getState() subsequently asks the channel to return the current state. If there is a current state (there may not be any other members in the group !), then true is returned. In this case, one of the subsequent receive() method invocations on the channel will return a SetStateEvent object which contains the current state. In this case, the caller sets its state to the one received from the channel.

Method receive() might return a GetStateEvent object, requesting the state of the member to be returned. In this case, a copy of the current state should be made and returned using JChannel.returnState() . It is important to a) synchronize access to the state when returning it since other accesses may modify it while it is being returned and b) make a copy of the state since other accesses after returning the state may still be able to modify it ! This is possible because the state is not immediately returned, but travels down the stack (in the same address space), and a reference to it could still alter it.

As an alternative to handling the GetStateEvent and SetStateEvent events, and calling Channel.returnState(), a Receiver could be used. The example above would look like this:

          class MyReceiver extends ReceiverAdapter {
              final Map m=new HashMap();
              public byte[] getState() {
                  // so nobody else can modify the map while we serialize it
                  synchronized(m) {
                      byte[] state=Util.objectToByteBuffer(m);
                      return state;
                  }
              }

              public void setState(byte[] state) {
                  synchronized(m) {
                      Map new_m=(Map)Util.objectFromByteBuffer(state);
                      m.clear();
                      m.addAll(new_m);
                  }
              }
          }
          // use default props (has to include STATE_TRANSFER)
          channel=new JChannel();
          channel.setReceiver(new MyReceiver());
          channel.connect("TestChannel");
          boolean rc=channel.getState(null, 5000);
      

In a group consisting of A,B and C, with D joining the group and calling Channel.getState(), the following sequence of callbacks happens:

Partial state transfer means that instead of transferring the entire state, we may want to transfer only a substate. For example, with HTTP session replication, a new node in a cluster may want to transfer only the state of a specific session, not all HTTP sessions. This can be done with either the pull or push model. The method to call would be Channel.getState(), including the ID of the substate (a string). In the pull model, GetStateEvent and SetStateEvent have an additional member, state_id, and in the push model, there are 2 additional getState() and setState() callbacks. The example below shows partial state transfer for the push model:

          class MyReceiver extends ExtendedReceiverAdapter {
              final Map m=new HashMap();

              public byte[] getState() {
                  return getState(null);
              }

              public byte[] getState(String substate_id) {
                  // so nobody can modify the map while we serialize it
                  synchronized(m) {
                      byte[] state=null;
                      if(substate_id == null) {
                          state=Util.objectToByteBuffer(m);
                      }
                      else {
                          Object value=m.get(substate_id);
                          if(value != null) {
                              return Util.objectToByteBuffer(value);
                          }
                      }
                      return state;
                  }
              }

              public void setState(byte[] state) {
                  setState(null, state);
              }

              public void setState(String substate_id, byte[] state) {
                  synchronized(m) {
                      if(substate_id != null) {
                          Object value=Util.objectFromByteBuffer(state);
                          m.put(substate_id, value);
                      }
                      else {
                          Map new_m=(Map)Util.objectFromByteBuffer(state);
                          m.clear();
                          m.addAll(new_m);
                      }
                  }
              }
          }

          // use default props (has to include pbcast.STATE_TRANSFER)
          channel=new JChannel();
          channel.setReceiver(new MyReceiver());
          channel.connect("TestChannel");
          boolean rc=channel.getState(null, "MyID", 5000);
      

The example shows that the Channel.getState() method specifies the ID of the substate, in this case "MyID". The getState(String substate_id) method checks whether the substate ID is not null, and returns the substate pertaining to the ID, or the entire state if the substate_id is null. The same goes for setting the substate: if setState(String substate_id, byte[] state) has a non-null substate_id, only that part of the current state will be overwritten, otherwise (if null) the entire state will be overwritten.

Streaming state transfer allows transfer of application (partial) state without having to load entire state into memory prior to sending it to a joining member. Streaming state transfer is especially useful if the state is very large (>1Gb), and use of regular state transfer would likely result in OutOfMemoryException. Streaming state transfer was introduced in JGroups 2.4. JGroups channel has to be configured with either regular or streaming state transfer. The JChannel API that invokes state transfer (i.e. JChannel.getState(long timeout, Address member)) remains the same.

Streaming state transfer, just as regular byte based state transfer, can be used in both pull and push mode. Similarly to the current getState and setState methods of org.jgroups.MessageListener, the application interested in streaming state transfer in a push mode would implement streaming getState method(s) by sending/writing state through a provided OutputStream reference and setState method(s) by receiving/reading state through a provided InputStream reference. In order to use streaming state transfer in a push mode, existing ExtendedMessageListener has been expanded to include additional four methods:

         public interface ExtendedMessageListener {

             /*non-streaming callback methods ommitted for clarity*/
 
             void getState(OutputStream ostream);
             void getState(String state_id, OutputStream ostream);
             void setState(InputStream istream);
             void setState(String state_id, InputStream istream);
          }
      

For a pull mode (when application uses channel.receive() to fetch events) two new event classes will be introduced:

These two events/classes are very similar to existing GetStateEvent and SetStateEvent but introduce a new field; StreamingGetStateEvent has an OutputStream and StreamingSetStateEvent has an InputStream.

The following code snippet demonstrates how to pull events from a channel, processing StreamingGetStateEvent and sending hypothetical state through a provided OutputStream reference. Handling of StreamingSetStateEvent is analogous to this example:

          ...
          Object obj=channel.receive(0);
          if(obj instanceof StreamingGetStateEvent) {
              StreamingGetStateEvent evt=(StreamingGetStateEvent)obj;
              OutputStream oos = null;
              try {
                  oos=new ObjectOutputStream(evt.getArg());
                  oos.writeObject(state);
                  oos.flush();
              }
              catch (Exception e) {}
              finally {
                  try {
                      oos.close();
                  }
                  catch (IOException e) {
                      System.err.println(e);
                  }
              }
          }                 
          

JGroups has a great flexibility with state transfer methodology by allowing application developers to implement both byte based and streaming based state transfers. Application can, for example, implement streaming and byte based state transfer callbacks and then interchange state transfer protocol in channel configuration to use either streaming or byte based state transfer. However, one cannot configure a channel with both state transfers at the same time and then in runtime choose which particular state transfer type to use.

Disconnecting from a channel is done using the following method:

          public void disconnect();
          

It will have no effect if the channel is already in the disconnected or closed state. If connected, it will remove itself from the group membership. This is done (transparently for a channel user) by sending a leave request to the current coordinator. The latter will subsequently remove the channel's address from its local view and send the new view to all remaining members.

After a successful disconnect, the channel will be in the unconnected state, and may subsequently be re-connected to.

To destroy a channel instance (destroy the associated protocol stack, and release all resources), method close() is used:

         public void close();
        

It moves the channel to the closed state, in which no further operations are allowed (most throw an exception when invoked on a closed channel). In this state, a channel instance is not considered used any longer by an application and -- when the reference to the instance is reset -- the channel essentially only lingers around until it is garbage collected by the Java runtime system.

This section shows sample code for using a PullPushAdapter. The example has been shortened for readability (error handling has been removed).

    public class PullPushTest implements MessageListener {
        Channel          channel;
        PullPushAdapter  adapter;
        byte[]           data="Hello world".getBytes();
        String           props; // fetch properties

        public void receive(Message msg) {
            System.out.println("Received msg: " + msg);
        }

        public void start() throws Exception {
            channel=new JChannel(props);
            channel.connect("PullPushTest");
            adapter=new PullPushAdapter(channel);
            adapter.setListener(this);

            for(int i=0; i < 10; i++) {
                System.out.println("Sending msg #" + i);
                channel.send(new Message(null, null, data));
                Thread.currentThread().sleep(1000);
            }
            adapter.stop();
            channel.close();
        }


        public static void main(String args[]) {
            try {
                new PullPushTest().start();
            }
            catch(Exception e) { /* error */ }
        }
    }
      

First a channel is created and connected to. Then an instance of PullPushAdapter is created with the channel as argument. The constructor of PullPushAdapter starts its own thread which continually reads on the channel. Then the MessageListener is set, which causes all messages received on the channel to be sent to receive(). Then a number of messages are sent via the channel to the entire group. As group messages are also received by the sender, the receive() method will be called every time a message is received. Finally the PullPushAdapter is stopped and the channel closed. Note that explicitly stopping the PullPushAdapter is not actually necessary, a closing the channel would cause the PullPushAdapter to terminate anyway.

Note that, compared to the pull-style example, push-style message reception is considerably easier (no separate thread management) and requires less code to program.

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

    public class MessageDispatcherTest implements RequestHandler {
        Channel            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, null, null, this);
            channel.connect("MessageDispatcherTestGroup");

            for(int i=0; i < 10; i++) {
                Util.sleep(100);
                System.out.println("Casting message #" + i);
                rsp_list=disp.castMessage(null,
                    new Message(null, null, new String("Number #" + i)),
                    GroupRequest.GET_ALL, 0);
                System.out.println("Responses:\n" +rsp_list);
            }
            channel.close();
            disp.stop();
        }

        public Object handle(Message msg) {
            System.out.println("handle(): " + msg);
            return new String("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.

Finally both the MessageDispatcher and channel are closed.

The code below shows an example:

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

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

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

            for(int i=0; i < 10; i++) {
                Util.sleep(100);
                rsp_list=disp.callRemoteMethods(null, "print",
                     new Integer(i), GroupRequest.GET_ALL, 0);
                System.out.println("Responses: " +rsp_list);
            }
            channel.close();
            disp.stop();
         }

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

Class RpcDispatcher defines method print() which will be called subsequently. The entry point start() method creates a channel and an RpcDispatcher which is layered on top. Method callRemoteMethods() then invokes the remote print() method in all group 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.

      public RspList callRemoteMethods(Collection<Address> dests, String method_name,
                                       Object[] args,Class[] types, RequestOptions options);
      public RspList callRemoteMethods(Collection<Address> dests, MethodCall method_call,
                                       RequestOptions options);
      public Object callRemoteMethod(Address dest, String method_name, Object[] args,
                                     Class[] types, RequestOptions options);
      public Object callRemoteMethod(Address dest, MethodCall call, RequestOptions options);
      

        RpcDispatcher disp;
        RequestOptions opts=new RequestOptions(Request.GET_ALL)
                                .setFlags(Message.NO_FC | Message.DONT_BUNDLE);
        Object val=disp.callRemoteMethod(target, method_call, opts);
      

        public NotifyingFuture<RspList> callRemoteMethodsWithFuture(Collection<Address> dests,
                                                   MethodCall method_call, RequestOptions options);
        public <T> NotifyingFuture<T> callRemoteMethodWithFuture(Address dest, MethodCall call,
                                                     RequestOptions options);
                

        NotifyingFuture<RspList> future=dispatcher.callRemoteMethodsWithFuture(...);
        future.setListener(new FutureListener() {
            void futureDone(Future<T> future) {
                System.out.println("result is " + future.get());
            }
        }
        );
      

The following scenario is susceptible to 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 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.:

                    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
                        }
                    }
                

UDP uses IP multicast for sending messages to all members of a group and UDP datagrams for unicast messages (sent to a single member). When started, it opens a unicast and multicast socket: the unicast socket is used to send/receive unicast messages, whereas the multicast socket sends/receives multicast messages. The channel's address will be the address and port number of the unicast socket.

              java org.jgroups.stack.GossipRouter -port 5555 -bindaddress localhost
          

            <UDP ip_mcast="false" mcast_addr="224.0.0.35" mcast_port="45566" ip_ttl="32"
                mcast_send_buf_size="150000" mcast_recv_buf_size="80000"/>
            <PING gossip_host="localhost" gossip_port="5555" gossip_refresh="15000"
                timeout="2000" num_initial_members="3"/>
        

TCP is a replacement of UDP as bottom layer in cases where IP Multicast based on UDP is not desired. This may be the case when operating over a WAN, where routers will discard IP MCAST. As a rule of thumb UDP is used as transport for LANs, whereas TCP is used for WANs.

The properties for a typical stack based on TCP might look like this (edited/protocols removed for brevity):

    <TCP start_port="7800" />
    <TCPPING timeout="3000"
             initial_hosts="${jgroups.tcpping.initial_hosts:localhost[7800],localhost[7801]}"
             port_range="1"
             num_initial_members="3"/>
    <VERIFY_SUSPECT timeout="1500"  />
    <pbcast.NAKACK
                   use_mcast_xmit="false" gc_lag="0"
                   retransmit_timeout="300,600,1200,2400,4800"
                   discard_delivered_msgs="true"/>
    <pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
                   max_bytes="400000"/>
    <pbcast.GMS print_local_addr="true" join_timeout="3000"
                shun="true"
                view_bundling="true"/>
      

Since TCP already offers some of the reliability guarantees that UDP doesn't, some protocols (e.g. FRAG and UNICAST) are not needed on top of TCP.

When using TCP, each message to the group is sent as multiple unicast messages (one to each member). Due to the fact that IP multicasting cannot be used to discover the initial members, another mechanism has to be used to find the initial membership. There are a number of alternatives:

The next two section illustrate the use of TCP with both TCPPING and TCPGOSSIP.

            <TCP start_port="7800" /> +
            <TCPPING initial_hosts="HostA[7800],HostB[7800]" port_range="5"
            timeout="3000" num_initial_members="3" />
        

              <TCP />
              <TCPGOSSIP initial_hosts="localhost[5555],localhost[5556]" gossip_refresh_rate="10000"
              num_initial_members="3" />
          

The architecture of the concurrent stack is shown in Figure 5.3, “The concurrent stack”. The changes were made entirely inside of the transport protocol (TP, with subclasses UDP, TCP and TCP_NIO). Therefore, to configure the concurrent stack, the user has to modify the config for (e.g.) UDP in the XML file.

Figure 5.3. The concurrent stack

The concurrent stack consists of 2 thread pools (java.util.concurrent.Executor): the out-of-band (OOB) thread pool and the regular thread pool. Packets are received by multicast or unicast receiver threads (UDP) or a ConnectionTable (TCP, TCP_NIO). Packets marked as OOB (with Message.setFlag(Message.OOB)) are dispatched to the OOB thread pool, and all other packets are dispatched to the regular thread pool.

When a thread pool is disabled, then we use the thread of the caller (e.g. multicast or unicast receiver threads or the ConnectionTable) to send the message up the stack and into the application. Otherwise, the packet will be processed by a thread from the thread pool, which sends the message up the stack. When all current threads are busy, another thread might be created, up to the maximum number of threads defined. Alternatively, the packet might get queued up until a thread becomes available.

The point of using a thread pool is that the receiver threads should only receive the packets and forward them to the thread pools for processing, because unmarshalling and processing is slower than simply receiving the message and can benefit from parallelization.

                        
                        <UDP
                                mcast_addr="228.10.10.10"
                                mcast_port="45588"

                                thread_pool.enabled="true"
                                thread_pool.min_threads="1"
                                thread_pool.max_threads="100"
                                thread_pool.keep_alive_time="20000"
                                thread_pool.queue_enabled="false"
                                thread_pool.queue_max_size="10"
                                thread_pool.rejection_policy="Run"

                                oob_thread_pool.enabled="true"
                                oob_thread_pool.min_threads="1"
                                oob_thread_pool.max_threads="4"
                                oob_thread_pool.keep_alive_time="30000"
                                oob_thread_pool.queue_enabled="true"
                                oob_thread_pool.queue_max_size="10"
                                oob_thread_pool.rejection_policy="Run"/>
                                
                    

By removing the 2 queues/protocol and the associated 2 threads, we effectively reduce the number of threads needed to handle a message, and thus context switching overhead. We also get clear and unambiguous semantics for Channel.send(): now, all messages are sent down the stack on the caller's thread and the send() call only returns once the message has been put on the network. In addition, an exception will only be propagated back to the caller if the message has not yet been placed in a retransmit buffer. Otherwise, JGroups simply logs the error message but keeps retransmitting the message. Therefore, if the caller gets an exception, the message should be re-sent.

On the receiving side, a message is handled by a thread pool, either the regular or OOB thread pool. Both thread pools can be completely eliminated, so that we can save even more threads and thus further reduce context switching. The point is that the developer is now able to control the threading behavior almost completely.

Up to version 2.5, all messages received were processed by a single thread, even if the messages were sent by different senders. For instance, if sender A sent messages 1,2 and 3, and B sent message 34 and 45, and if A's messages were all received first, then B's messages 34 and 35 could only be processed after messages 1-3 from A were processed !

Now, we can process messages from different senders in parallel, e.g. messages 1, 2 and 3 from A can be processed by one thread from the thread pool and messages 34 and 35 from B can be processed on a different thread.

As a result, we get a speedup of almost N for a cluster of N if every node is sending messages and we configure the thread pool to have at least N threads. There is actually a unit test (ConcurrentStackTest.java) which demonstrates this.

In the previous paragraph, we showed how the concurrent stack delivers messages from different senders concurrently. But all (non-OOB) messages from the same sender P are delivered in the order in which P sent them. However, this is not good enough for certain types of applications.

Consider the case of an application which replicates HTTP sessions. If we have sessions X, Y and Z, then updates to these sessions are delivered in the order in which there were performed, e.g. X1, X2, X3, Y1, Z1, Z2, Z3, Y2, Y3, X4. This means that update Y1 has to wait until updates X1-3 have been delivered. If these updates take some time, e.g. spent in lock acquisition or deserialization, then all subsequent messages are delayed by the sum of the times taken by the messages ahead of them in the delivery order.

However, in most cases, updates to different web sessions should be completely unrelated, so they could be delivered concurrently. For instance, a modification to session X should not have any effect on session Y, therefore updates to X, Y and Z can be delivered concurrently.

One solution to this is out-of-band (OOB) messages (see next paragraph). However, OOB messages do not guarantee ordering, so updates X1-3 could be delivered as X1, X3, X2. If this is not wanted, but messages pertaining to a given web session should all be delivered concurrently between sessions, but ordered within a given session, then we can resort to scoped messages.

Scoped messages apply only to regular (non-OOB) messages, and are delivered concurrently between scopes, but ordered within a given scope. For example, if we used the sessions above (e.g. the jsessionid) as scopes, then the delivery could be as follows ('->' means sequential, '||' means concurrent):

                    X1 -> X2 -> X3 -> X4 || Y1 -> Y2 -> Y3 || Z1 -> Z2 -> Z3
                

This means that all updates to X are delivered in parallel to updates to Y and updates to Z. However, within a given scope, updates are delivered in the order in which they were performed, so X1 is delivered before X2, which is deliverd before X3 and so on.

Taking the above example, using scoped messages, update Y1 does not have to wait for updates X1-3 to complete, but is processed immediately.

To set the scope of a message, use method Message.setScope(short).

Scopes are implemented in a separate protocol called Section 7.13.3, “SCOPE”. This protocol has to be placed somewhere above ordering protocols like UNICAST or NAKACK (or SEQUENCER for that matter).

OOB messages completely ignore any ordering constraints the stack might have. Any message marked as OOB will be processed by the OOB thread pool. This is necessary in cases where we don't want the message processing to wait until all other messages from the same sender have been processed, e.g. in the heartbeat case: if sender P sends 5 messages and then a response to a heartbeat request received from some other node, then the time taken to process P's 5 messages might take longer than the heartbeat timeout, so that P might get falsely suspected ! However, if the heartbeat response is marked as OOB, then it will get processed by the OOB thread pool and therefore might be concurrent to its previously sent 5 messages and not trigger a false suspicion.

The 2 unit tests UNICAST_OOB_Test and NAKACK_OOB_Test demonstrate how OOB messages influence the ordering, for both unicast and multicast messages.

In 2.7, there are 3 thread pools and 4 thread factories in TP:

In 2.7, the default and OOB thread pools can be shared between instances running inside the same JVM. The advantage here is that multiple channels running within the same JVM can pool (and therefore save) threads. The disadvantage is that thread naming will not show to which channel instance an incoming thread belongs to.

Note that we can not just shared thread pools between JChannels within the same JVM, but we can also share entire transports. For details see Section 5.2, “The shared transport: sharing a transport between multiple channels in a JVM”.

Note that in 2.8, shunning has been removed, so the sections below only apply to versions up to 2.7.

                channel.setOpt(Channel.AUTO_RECONNECT, Boolean.TRUE);
            

To enable shunning, set FD.shun and GMS.shun to true.

                <FD timeout="2000" max_tries="3" shun="false"/>
                ...
                <pbcast.GMS join_timeout="3000" shun="false"/>
            

JGroups creates all of its sockets through a SocketFactory, which is located in the transport (TP) or TP.ProtocolAdapter (in a shared transport). The factory has methods to create sockets (Socket, ServerSocket, DatagramSocket and MulticastSocket) ^[12], closen sockets and list all open sockets. Every socket creation method has a service name, which could be for example "jgroups.fd_sock.srv_sock". The service name is used to look up a port (e.g. in a config file) and create the correct socket.

To provide one's own socket factory, the following has to be done: if we have a non-shared transport, the code below creates a SocketFactory implementation and sets it in the transport:

                JChannel ch;
                MySocketFactory factory; // e.g. extends DefaultSocketFactory
                ch=new JChannel("config.xml");
                ch.setSocketFactory(new MySocketFactory());
                ch.connect("demo");
            

If a shared transport is used, then we have to set 2 socket factories: 1 in the shared transport and one in the TP.ProtocolAdapter:

                JChannel c1=new JChannel("config.xml"), c2=new JChannel("config.xml");

                TP transport=c1.getProtocolStack().getTransport();
                transport.setSocketFactory(new MySocketFactory("transport"));

                c1.setSocketFactory(new MySocketFactory("first-cluster"));
                c2.setSocketFactory(new MySocketFactory("second-cluster"));

                c1.connect("first-cluster");
                c2.connect("second-cluster");
            

First, we grab one of the channels to fetch the transport and set a SocketFactory in it. Then we set one SocketFactory per channel that resides on the shared transport. When JChannel.connect() is called, the SocketFactory will be set in TP.ProtocolAdapter.

The application has to merge the substates from the various subgroups ({A,B,C} and {D,E}) back into one single state for {A,B,C,D,E}. This task has to be done by the application because JGroups knows nothing about the application state, other than it is a byte buffer.

If the in-memory state is backed by a database, then the solution is easy: simply discard the in-memory state and fetch it (eagerly or lazily) from the DB again. This of course assumes that the members of the 2 subgroups were able to write their changes to the DB. However, this is often not the case, as connectivity to the DB might have been severed by the network partition.

Another solution could involve tagging the state with time stamps. On merging, we could compare the time stamps for the substates and let the substate with the more recent time stamps win.

Yet another solution could increase a counter for a state each time the state has been modified. The state with the highest counter wins.

Again, the merging of state can only be done by the application. Whatever algorithm is picked to merge state, it has to be deterministic.

In certain situations, we can avoid having multiple subgroups where every subgroup is able to make progress, and on merging having to discard state of the non-primary partitions.

If we have a fixed membership, e.g. the cluster always consists of 5 nodes, then we can run code on a view reception that determines the primary partition. This code

The algorithm is shown in pseudo code below:

                    On initialization:
                        - Mark the node as read-only
                    
                    On view change V:
                        - If V has >= N members:
                            - If not read-write: get state from coord and switch to read-write
                        - Else: switch to read-only
                

Of course, the above mechanism requires that at least 3 nodes are up at any given time, so upgrades have to be done in a staggered way, taking only one node down at a time. In the worst case, however, this mechanism leaves the cluster read-only and notifies a system admin, who can fix the issue. This is still better than shutting the entire cluster down.

When we have a chatty protocol, scaling to a large number of nodes might be a problem: too many messages are sent and - because they are generated in addition to the regular traffic - this can have a negative impact on the cluster. A possible impact is that more of the regular messages are dropped, and have to be retransmitted, which impacts performance. Or heartbeats are dropped, leading to false suspicions. So while the negative effects of chatty protocols may not be seen in small clusters, they will be seen in large clusters !

RELAY presents a global view to the application, e.g. a view received by nodes could be {D,E,F,A,B,C}. This view is the same on all nodes, and a global view is generated by taking the two local views, e.g. A|5 {A,B,C} and D|2 {D,E,F}, comparing the coordinators' addresses (the UUIDs for A and D) and concatenating the views into a list. So if D's UUID is greater than A's UUID, we first add D's members into the global view ({D,E,F}), and then A's members.

Therefore, we'll always see all of A's members, followed by all of D's members, or the other way round.

To see which nodes are local and which ones remote, we can iterate through the addresses (PayloadUUID) and use the site (PayloadUUID.getPayload()) name to for example differentiate between "nyc" and "sfo".

To setup a relay, we need essentially 3 XML configuration files: 2 to configure the local clusters and 1 for the bridge.

To configure the first local cluster, we can copy udp.xml from the JGroups distribution and add RELAY on top of it: <RELAY bridge_props="/home/bela/tcp.xml" />. Let's say we call this config relay.xml.

The second local cluster can be configured by copying relay.xml to relay2.xml. Then change the mcast_addr and/or mcast_port, so we actually have 2 different cluster in case we run instances of both clusters in the same network. Of course, if the nodes of one cluster are run in a different network from the nodes of the other cluster, and they cannot talk to each other, then we can simply use the same configuration.

The 'site' property needs to be configured in relay.xml and relay2.xml, and it has to be different. For example, relay.xml could use site="nyc" and relay2.xml could use site="sfo".

The bridge is configured by taking the stock tcp.xml and making sure both local clusters can see each other through TCP.

It takes 6 rounds to broadcast m1 and m2 to the cluster.

In round 1, A send m1 to B.

In round 2, A sends m2 to B, but B also forwards m1 (received in round 1) to C.

In round 3, A is done. B forwards m2 to C and C forwards m1 to D (in parallel, denoted by '||').

In round 4, C forwards m2 to D.

Let's take a look at this in terms of switch usage: in the N-1 approach, A can only send 125MB/sec, no matter how many members there are in the cluster, so it is constrained by the link capacity to the switch. (Note that A can also receive 125MB/sec in parallel with today's full duplex links).

So the link between A and the switch gets hot.

In the daisychaining approach, link usage is more even: if we look for example at round 2, A sending to B and B sending to C uses 2 different links, so there are no constraints regarding capacity of a link. The same goes for B sending to C and C sending to D.

In terms of rounds, the daisy chaining approach uses X + (N-2) rounds, so for a cluster size of 10 and broadcasting 10 messages, it requires only 18 rounds, compared to 90 for the N-1 approach !

To measure performance of DAISYCHAIN, a performance test (test.Perf) was run, with 4 nodes connected to a 1 GB switch; and every node sending 1 million 8K messages, for a total of 32GB received by every node. The config used was tcp.xml.

The N-1 approach yielded a throughput of 73 MB/node/sec, and the daisy chaining approach 107MB/node/sec !

DAISYCHAIN can be placed directly on top of the transport, regardless of whether it is UDP or TCP, e.g.

                    <TCP .../>
                    <DAISYCHAIN .../>
                    <TCPPING .../>
                

FILE_PING can be used instead of GossipRouter in cases where no external process is desired.

JDBC_PING is an alternative to S3_PING by using Amazon RDS instead of S3.

BPING uses UDP broadcasts to discover other nodes. The default broadcast address (dest) is 255.255.255.255, and should be replaced with a subnet specific broadcast, e.g. 192.168.1.255.

S3_PING is primarily meant to be used on Amazon EC2 where multicast traffic is not allowed and no external process (GossipRouter) is desired. When Amazon RDS is preferred over S3, or if a shared database is used, an alternative is to use JDBC_PING.

Failure detection based on heartbeat messages. If reply is not received without timeout ms, max_tries times, a member is declared suspected, and will be excluded by GMS

Each member send a message containing a "FD" - HEARTBEAT header to its neighbor to the right (identified by the ping_dest address). The heartbeats are sent by the inner class Monitor. When the neighbor receives the HEARTBEAT, it replies with a message containing a "FD" - HEARTBEAT_ACK header. The first member watches for "FD" - HEARTBEAT_ACK replies from its neigbor. For each received reply, it resets the last_ack timestamp (sets it to current time) and num_tries counter (sets it to 0). The same Monitor instance that sends heartbeats whatches the difference between current time and last_ack. If this difference grows over timeout, the Monitor cycles several more times (until max_tries) is reached) and then sends a SUSPECT message for the neighbor's address. The SUSPECT message is sent down the stack, is addressed to all members, and is as a regular message with a FdHeader.SUSPECT header.

Failure detection based on simple heartbeat protocol. Every member periodically multicasts a heartbeat. Every member also maintains a table of all members (minus itself). When data or a heartbeat from P are received, we reset the timestamp for P to the current time. Periodically, we check for expired members, and suspect those.

Example: <FD_ALL interval="3000" timeout="10000"/>

In the example above, we send a heartbeat every 3 seconds and suspect members if we haven't received a heartbeat (or traffic) for more than 10 seconds. Note that since we check the timestamps every 'interval' milliseconds, we will suspect a member after roughly 4 * 3s == 12 seconds. If we set the timeout to 8500, then we would suspect a member after 3 * 3 secs == 9 seconds.

Failure detection protocol based on a ring of TCP sockets created between group members. Each member in a group connects to its neighbor (last member connects to first) thus forming a ring. Member B is suspected when its neighbor A detects abnormally closed TCP socket (presumably due to a node B crash). However, if a member B is about to leave gracefully, it lets its neighbor A know, so that it does not become suspected.

If you are using a multi NIC machine note that JGroups versions prior to 2.2.8 have FD_SOCK implementation that does not assume this possibility. Therefore JVM can possibly select NIC unreachable to its neighbor and setup FD_SOCK server socket on it. Neighbor would be unable to connect to that server socket thus resulting in immediate suspecting of a member. Suspected member is kicked out of the group, tries to rejoin, and thus goes into join/leave loop. JGroups version 2.2.8 introduces srv_sock_bind_addr property so you can specify network interface where FD_SOCK TCP server socket should be bound. This network interface is most likely the same interface used for other JGroups traffic. JGroups versions 2.2.9 and newer consult bind.address system property or you can specify network interface directly as FD_SOCK bind_addr property.

NAKACK provides reliable delivery and FIFO (= First In First Out) properties for messages sent to all nodes in a cluster.

Reliable delivery means that no message sent by a sender will ever be lost, as all messages are numbered with sequence numbers (by sender) and retransmission requests are sent to the sender of a message^[14] if that sequence number is not received.

FIFO order means that all messages from a given sender are received in exactly the order in which they were sent.

UNICAST provides reliable delivery and FIFO (= First In First Out) properties for point-to-point messages between one sender and one receiver.

Reliable delivery means that no message sent by a sender will ever be lost, as all messages are numbered with sequence numbers (by sender) and retransmission requests are sent to the sender of a message^[15] if that sequence number is not received.

FIFO order means that all messages from a given sender are received in exactly the order in which they were sent.

On top of a reliable transport, such as TCP, UNICAST is not really needed. However, concurrent delivery of messages from the same sender is prevented by UNICAST by acquiring a lock on the sender's retransmission table, so unless concurrent delivery is desired, UNICAST should not be removed from the stack even if TCP is used.

UNICAST2 provides lossless, ordered, communication between 2 members. Contrary to UNICAST, it uses negative acks (similar to NAKACK) rather than positive acks. This reduces the communication overhead required for sending an ack for every message.

SEQUENCER provider total order for multicast (=group) messages by forwarding messages to the current coordinator, which then sends the messages to the cluster on behalf of the original sender. Because it is always the same sender (whose messages are delivered in FIFO order), a global (or total) order is established.

Sending members add every forwarded message M to a buffer and remove M when they receive it. Should the current coordinator crash, all buffered messages are forwarded to the new coordinator.

Note that retransmissions go to the original sender, not to the coordinator.

FC uses a credit based system, where each sender has max_credits credits and decrements them whenever a message is sent. The sender blocks when the credits fall below 0, and only resumes sending messages when it receives a replenishment message from the receivers.

The receivers maintain a table of credits for all senders and decrement the given sender's credits as well, when a message is received.

When a sender's credits drops below a threshold, the receiver will send a replenishment message to the sender. The threshold is defined by min_bytes or min_threshold.

A simplified version of FC. FC can actually still overrun receivers when the transport's latency is very small. SFC is a simple flow control protocol for group (= multipoint) messages.

Every sender has max_credits bytes for sending multicast messages to the group.

Every multicast message (we don't consider unicast messages) decrements max_credits by its size. When max_credits falls below 0, the sender asks all receivers for new credits and blocks until *all* credits have been received from all members.

When the receiver receives a credit request, it checks whether it has received max_credits bytes from the requester since the last credit request. If yes, it sends new credits to the requester and resets the max_credits for the requester. Else, it takes a note of the credit request from P and - when max_credits bytes have finally been received from P - it sends the credits to P and resets max_credits for P.

The maximum amount of memory for received messages is therefore <number of senders> * max_credits.

The relationship with STABLE is as follows: when a member Q is slow, it will prevent STABLE from collecting messages above the ones seen by Q (everybody else has seen more messages). However, because Q will *not* send credits back to the senders until it has processed all messages worth max_credits bytes, the senders will block. This in turn allows STABLE to progress and eventually garbage collect most messages from all senders. Therefore, SFC and STABLE complement each other, with SFC blocking senders so that STABLE can catch up.

SFC is currently experimental, we recommend to use MFC and UFC (see below) instead.

In 2.10, FC was separated into MFC (Multicast Flow Control) and Unicast Flow Control (UFC). The reason was that multicast flow control should not be impeded by unicast flow control, and vice versa. Also, performance for the separate implementations could be increased, plus they can be individually omitted. For example, if no unicast flow control is needed, UFC can be left out of the stack configuration.

Flushing forces group members to send all their pending messages prior to a certain event. The process of flushing acquiesces the cluster so that state transfer or a join can be done. It is also called the stop-the-world model as nobody will be able to send messages while a flush is in process. Flush is used:

FLUSH is designed as another protocol positioned just below the channel, e.g. above STATE_TRANSFER and FC. STATE_TRANSFER and GMS protocol request flush by sending a SUSPEND event up the stack, where it is handled by the FLUSH protcol. The SUSPEND_OK ack sent back by the FLUSH protocol let's the caller know that the flush has completed. When done (e.g. view was installed or state transferred), the protocol sends up a RESUME event, which will allow everyone in the cluster to resume sending.

Channel can be notified that FLUSH phase has been started by turning channel block option on. By default it is turned off. If channel blocking is turned on FLUSH notifies application layer that channel has been blocked by sending EVENT.BLOCK event. Channel responds by sending EVENT.BLOCK_OK event down to FLUSH protocol. We recommend turning on channel block notification only if channel is used in push mode. In push mode application that uses channel can perform block logic by implementing MembershipListener.block() callback method.

As discussed in Section 5.4.4, “Scopes: concurrent message delivery for messages from the same sender”, the SCOPE protocol is used to deliver updates to different scopes concurrently. It has to be placed somewhere above UNICAST and NAKACK.

SCOPE has a separate thread pool. The reason why the default thread pool from the transport wasn't used is that the default thread pool has a different purpose. For example, it can use a queue to which all incoming messages are added, which would defy the purpose of concurrent delivery in SCOPE. As a matter of fact, using a queue would most likely delay messages get sent up into SCOPE !

Also, the default pool's rejection policy might not be "run", so the SCOPE implementation would have to catch rejection exceptions and engage in a retry protocol, which is complex and wastes resources.

The configuration of the thread pool is shown below. If you expect concurrent messages to N different scopes, then the max pool size would ideally be set to N. However, in most cases, this is not necessary as (a) the messages might not be to different scopes or (b) not all N scopes might get messages at the same time. So even if the max pool size is a bit smaller, the cost of this is slight delays, in the sense that a message for scope Y might wait until the thread processing message for scope X is available.

To remove unused scopes, an expiry policy is provided: expiration_time is the number of milliseconds after which an idle scope is removed. An idle scope is a scope which hasn't seen any messages for expiration_time milliseconds. The expiration_interval value defines the number of milliseconds at which the expiry task runs. Setting both values to 0 disables expiration; it would then have to be done manually (see Section 5.4.4, “Scopes: concurrent message delivery for messages from the same sender” for details).

RELAY bridges traffic between seperate clusters, see Section 5.9, “Bridging between remote clusters” for details.

STOMP is a JGroups protocol which implements the STOMP protocol. Currently (as of Nov 2010), transactions and acks are not implemented.

The location of a STOMP protocol in a stack is shown in Figure 7.1, “STOMP in a protocol stack”.

Figure 7.1. STOMP in a protocol stack

The STOMP protocol should be near the top of the stack.

A STOMP instance listens on a TCP socket for client connections. The port and bind address of the server socket can be defined via properties.

A client can send SUBSCRIBE commands for various destinations. When a SEND for a given destination is received, STOMP adds a header to the message and broadcasts it to all cluster nodes. Every node then in turn forwards the message to all of its connected clients which have subscribed to the same destination. When a destination is not given, STOMP simply forwards the message to all connected clients.

Traffic can be generated by clients and by servers. In the latter case, we could for example have code executing in the address space of a JGroups (server) node. In the former case, clients use the SEND command to send messages to a JGroups server and receive messages via the MESSAGE command. If there is code on the server which generates messages, it is important that both client and server code agree on a marshalling format, e.g. JSON, so that they understand each other's messages.

Clients can be written in any language, as long as they understand the STOMP protocol. Note that the JGroups STOMP protocol implementation sends additional information (e.g. INFO) to clients; non-JGroups STOMP clients should simply ignore them.

JGroups comes with a STOMP client (org.jgroups.client.StompConnection) and a demo (StompDraw). Both need to be started with the address and port of a JGroups cluster node. Once they have been started, the JGroups STOMP protocol will notify clients of cluster changes, which is needed so client can failover to another JGroups server node when a node is shut down. E.g. when a client connects to C, after connection, it'll get a list of endpoints (e.g. A,B,C,D). When C is terminated, or crashes, the client automatically reconnects to any of the remaining nodes, e.g. A, B, or D. When this happens, a client is also re-subscribed to the destinations it registered for.

The JGroups STOMP protocol can be used when we have clients, which are either not in the same network segment as the JGroups server nodes, or which don't want to become full-blown JGroups server nodes. Figure 7.2, “STOMP architecture” shows a typical setup.

Figure 7.2. STOMP architecture

There are 4 nodes in a cluster. Say the cluster is in a LAN, and communication is via IP multicasting (UDP as transport). We now have clients which do not want to be part of the cluster themselves, e.g. because they're in a different geographic location (and we don't want to switch the main cluster to TCP), or because clients are frequently started and stopped, and therefore the cost of startup and joining wouldn't be amortized over the lifetime of a client. Another reason could be that clients are written in a different language, or perhaps, we don't want a large cluster, which could be the case if we for example have 10 JGroups server nodes and 1000 clients connected to them.

In the example, we see 9 clients connected to every JGroups cluster node. If a client connected to node A sends a message to destination /topics/chat, then the message is multicast from node A to all other nodes (B, C and D). Every node then forwards the message to those clients which have previously subscribed to /topics/chat.

When node A crashes (or leaves) the JGroups STOMP clients (org.jgroups.client.StompConnection) simply pick another server node and connect to it.

The properties for STOMP are shown below:

The DAISYCHAIN protocol is discussed in Section 5.10, “Daisychaining”.

RATE_LIMITER can be used to set a limit on the data sent per time unit. When sending data, only max_bytes can be sent per time_period milliseconds. E.g. if max_bytes="50M" and time_period="1000", then a sender can only send 50MBytes / sec max.

There are currently 2 locking protocols: org.jgroups.protocols.CENTRAL_LOCK and org.jgroups.protocols.PEER_LOCK.

CENTRAL_EXECUTOR is an implementation of Executing which is needed by the ExecutionService.