stable.java

JGRoups源码

// $Id: STABLE.java,v 1.11 2005/08/11 12:43:47 belaban Exp $

package org.jgroups.protocols;


import org.jgroups.*;
import org.jgroups.blocks.GroupRequest;
import org.jgroups.blocks.MethodCall;
import org.jgroups.stack.RpcProtocol;
import org.jgroups.util.TimeScheduler;
import org.jgroups.util.Util;

import java.util.Properties;
import java.util.Vector;


/**
 * Computes the broadcast messages that are stable; i.e., that have been received
 * by all members. Sends STABLE events up the stack when this is the case.
 * Uses a probabilistic scheme to do so, as described in:<br>
 * GSGC: An Efficient Gossip-Style Garbage Collection Scheme for Scalable
 * Reliable Multicast, K. Guo et al., 1997.
 * <p>
 * The only difference is that instead of using counters for an estimation of
 * messages received from each member, we retrieve this actual information
 * from the NAKACK layer (which must be present for the STABLE protocol to
 * work).
 * <p>
 * Note: the the <tt>Event.MSG</tt> call path path must be as lightweight as
 * possible. It should not request any lock for which there is a high
 * contention and/or long delay.
 * <p>
 * <pre>
 * Changes(igeorg - 2.VI.2001):
 * i. Thread-safety (in RPC calls most notably on the lines of Gianluca
 * Collot's bugfix)
 * ii. All slow calls (RPCs, seqnos requests, etc.) placed outside locks
 * iii. Removed redundant initialization in adaptation to a higher round
 * iv. heard_from[this meber] is always set to true on every new round
 * (i.e. on every stability bcast).
 * v. Replaced gossip thread with <tt>TimeScheduler.Task</tt>
 * </pre>
 * <p>
 * [[[ TODO(igeorg - 2.VI.2001)
 * i. Faster stability convergence by better selection of gossip subsets
 * (replace Util.pickSubset()).
 * ii. Special mutex on the <tt>Event.MSG</tt> call path. I.e. remove
 * <tt>synchronized(this)</t>> with e.g. <tt>synchronized(msg_mutex)</tt>.
 * ]] TODO
 */
public class STABLE extends RpcProtocol {
    /** The protocol name */
    private static final String PROT_NAME="STABLE";

    /** Default subgroup size for gossiping expressed as percentage overthe group's size */
    private static final double SUBSET_SIZE=0.1;

    /** Default max number of msgs to wait for before sending gossip */
    private static final int GOSSIP_MSG_INTERVAL=100;

    /** Default max time to wait before sending gossip (ms) */
    private static final int GOSSIP_INTERVAL=10000;

    private Address local_addr=null;
    private ViewId vid=null;
    private final Vector mbrs=new Vector(11);

    /** gossip round */
    private long round=1;

    /** highest seqno received for each member (corresponds to membership) */
    private long[] seqnos=new long[0];

    /** Array of members from which we have received a gossip in the current round */
    private boolean[] heard_from=new boolean[0];

    /** Percentage of members to which gossip is sent (parameterizable by user) */
    private double subset=SUBSET_SIZE;

    /** The gossiping task scheduler */
    private TimeScheduler sched=null;

    private Task gossip_task;

    /** wait for n messages until sending gossip ... */
    private int max_msgs=GOSSIP_MSG_INTERVAL;

    /** ... or until max_wait_time has elapsed, whichever comes first */
    private long max_wait_time=GOSSIP_INTERVAL;

    /** Current number of msgs left to be received before sending a gossip */
    private long num_msgs=max_msgs;

    /** mutex for interacting with NAKACK layer (GET_MSGS_RECVD) */
    private final Object highest_seqnos_mutex=new Object();
    
    /** Time to wait for a reply from NAKACK layer (GET_MSGS_RECVD) */
    private long highest_seqnos_timeout=4000;


    /**
     * @return this protocol name
     */
    public String getName() {
        return (PROT_NAME);
    }


    /**
     * The events expected to be handled from some layer above:
     * <ul>
     * <li>
     * GET_MSGS_RECEIVED: NAKACK layer
     * </li>
     * </ul>
     * @return a list of events expected by to be handled from some layer
     * above
     */
    public Vector requiredUpServices() {
        Vector retval=new Vector(1);
        retval.addElement(new Integer(Event.GET_MSGS_RECEIVED));
        return retval;
    }

    /**
     * Set the parameters for this layer.
     *
     * <ul>
     * <li>
     * <i>subset</i>: the percentage of the group'size to which the
     * msgs_seen_so_far gossip is sent periodically.</li>
     * <li>
     * <i>max_msgs</i>: the max number of msgs to wait for between two
     * consecutive gossipings.</li>
     * <li>
     * <i>max_wait_time</i>: the max time to wait for between two consecutive
     * gossipings.</li>
     * <li>
     * <i>highest_seqno_timeout</i>: time to wait to receive from NAKACK
     * the array of highest deliverable seqnos
     * </li>
     * </ul>
     *
     * @param props the list of parameters
     */
    public boolean setProperties(Properties props) {
        String str;

        super.setProperties(props);
        str=props.getProperty("subset");
        if(str != null) {
            subset=Float.parseFloat(str);
            props.remove("subset");
        }

        str=props.getProperty("max_msgs");
        if(str != null) {
            num_msgs=max_msgs=Integer.parseInt(str);
            if(max_msgs <= 1) {
                if(log.isFatalEnabled()) log.fatal("value for 'max_msgs' must be greater than 1 !");
                return false;
            }
            props.remove("max_msgs");
        }

        str=props.getProperty("max_wait_time");
        if(str != null) {
            max_wait_time=Long.parseLong(str);
            props.remove("max_wait_time");
        }

        str=props.getProperty("highest_seqnos_timeout");
        if(str != null) {
            highest_seqnos_timeout=Long.parseLong(str);
            props.remove("highest_seqnos_timeout");
        }

        if(props.size() > 0) {
            log.error("STABLE.setProperties(): these properties are not recognized: " + props);

            return false;
        }
        return true;
    }


    /**
     * Start the layer:
     * i. Set the gossip task scheduler
     * ii. Reset the layer's state.
     * iii. Start the gossiping task
     */
    public void start() throws Exception {
        TimeScheduler timer;

        super.start();
        timer=stack != null ? stack.timer : null;
        if(timer == null)
            throw new Exception("STABLE.start(): timer is null");

        sched=timer;

        // we use only asynchronous method invocations...
        if(_corr != null)
            _corr.setDeadlockDetection(false);
        initialize();
        startGossip();
    }


    /**
     * Stop scheduling the gossip task
     */
    public void stop() {
        super.stop();
        synchronized(this) {
            if(gossip_task != null)
                gossip_task.cancel();
            gossip_task=null;
        }
    }


    /* ------------------------- Request handler methods ------------------ */

    /**
     * Contains the highest sequence numbers as seen by <code>sender</code>
     *
     * @param view_id The view ID in which the gossip was sent. Must be the
     * same as ours, otherwise it is discarded
     *
     * @param gossip_round The round in which the gossip was sent
     *
     * @param gossip_seqnos A vector with the highest sequence numbers as
     * seen by <code>sender</code>
     *
     * @param heard The sender's <code>heard_from</code> array. This allows
     * us to minimize the gossip msgs for a given round as a member does not
     * have to receive gossip msgs from each member, but members pass gossips
     * they've received from others on in their own gossips. E.g. when a
     * member P (of group {P,Q,R}) receives a gossip from R, its own gossip
     * to Q might be {R,P}. Q, who hasn't received a gossip from R, will not
     * need to receive it anymore as it is already sent by P. This simple
     * scheme reduces the number of gossip messages needed.
     *
     * @param sender The sender of the gossip message (obviously :-))
     */
    public void gossip(ViewId view_id, long gossip_round,
                       long[] gossip_seqnos, boolean[] heard, Object sender) {
        Object[] params;
        MethodCall call;

        synchronized(this) {

                if(log.isInfoEnabled()) log.info("sender=" + sender + ", round=" + gossip_round + ", seqnos=" +
                        Util.array2String(gossip_seqnos) + ", heard=" +
                        Util.array2String(heard));
            if(vid == null || view_id == null || !vid.equals(view_id)) {

                    if(log.isInfoEnabled()) log.info("view ID s are different (" + vid + " != " + view_id +
                            "). Discarding gossip received");
                return;
            }
            if(gossip_round < this.round) {

                    if(log.isInfoEnabled()) log.info("received a gossip from a previous round (" +
                            gossip_round + "); my round is " + round +
                            ". Discarding gossip");
                return;
            }
            if(gossip_seqnos == null || seqnos == null ||
                    seqnos.length != gossip_seqnos.length) {

                    if(warn) log.warn("size of seqnos and gossip_seqnos are not equal ! " +
                            "Discarding gossip");
                return;
            }

            // (1) If round greater than local round:
            // i. Adjust the local to the received round
            //
            // (2)
            // i. local_seqnos = arrayMin(local_seqnos, gossip_seqnos)
            // ii. local_heard = arrayMax(local_heard, gossip_heard)
            // iii. If heard from all, bcast our seqnos (stability vector)
            if(round == gossip_round) {
                update(sender, gossip_seqnos, heard);
            }
            else if(round < gossip_round) {

                    if(log.isInfoEnabled()) log.info("received a gossip from a higher round (" +
                            gossip_round + "); adopting my round (" + round +
                            ") to " + gossip_round);
                round=gossip_round;
                set(sender, gossip_seqnos, heard_from);
            }

             if(log.isInfoEnabled()) log.info("heard_from=" + Util.array2String(heard_from));
            if(!heardFromAll())
                return;

            params=new Object[]{
                vid.clone(),
                new Long(gossip_round),
                seqnos.clone(),
                local_addr};
        } // synchronized(this)

        call=new MethodCall("stability", params, 
            new String[] {ViewId.class.getName(), long.class.getName(), long[].class.getName(), Object.class.getName()});
        callRemoteMethods(null, call, GroupRequest.GET_NONE, 0);
    }


    /**
     * Contains the highest message sequence numbers (for each member) that
     * can safely be deleted (because they have been seen by all members).
     */
    public void stability(ViewId view_id, long gossip_round, long[] stability_vector, Object sender) {
        // i. Proceed to the next round; init the heard from list
        // ii. Send up the stability vector
        // iii. get a fresh copy of the highest deliverable seqnos
        synchronized(this) {

                if(log.isInfoEnabled()) log.info("sender=" + sender + ", round=" + gossip_round + ", vector=" +
                        Util.array2String(stability_vector) + ')');
            if(vid == null || view_id == null || !vid.equals(view_id)) {

                    if(log.isInfoEnabled()) log.info("view ID s are different (" + vid + " != " + view_id +
                            "). Discarding gossip received");
                return;
            }

            if(round > gossip_round)
                return;
            round=gossip_round + 1;
            for(int i=0; i < heard_from.length; i++)
                heard_from[i]=false;
        }
        heard_from[mbrs.indexOf(local_addr)]=true;

        passUp(new Event(Event.STABLE, stability_vector));
        getHighestSeqnos();
    }

    /* --------------------- End of Request handler methods --------------- */

    /**
     * <b>Callback</b>. Called by superclass when event may be handled.
     * <p>
     * <b>Do not use <code>PassUp</code> in this method as the event is passed
     * up by default by the superclass after this method returns !</b>
     *
     * @return boolean Defaults to true. If false, event will not be passed
     * up the stack.
     */
    public boolean handleUpEvent(Event evt) {
        switch(evt.getType()) {
            case Event.MSG:
                if(!upMsg(evt))
                    return (false);
                break;
            case Event.SET_LOCAL_ADDRESS:
                local_addr=(Address)evt.getArg();
                break;
        }

        return true;
    }


    /**
     * <b>Callback</b>. Called by superclass when event may be handled.
     * <p>
     * <b>Do not use <code>PassDown</code> in this method as the event is