<commit_before>package org.jgroups.protocols.pbcast;


import org.jgroups.*;
import org.jgroups.annotations.MBean;
import org.jgroups.annotations.ManagedAttribute;
import org.jgroups.annotations.ManagedOperation;
import org.jgroups.annotations.Property;
import org.jgroups.conf.PropertyConverters;
import org.jgroups.protocols.TP;
import org.jgroups.protocols.pbcast.GmsImpl.Request;
import org.jgroups.stack.DiagnosticsHandler;
import org.jgroups.stack.MembershipChangePolicy;
import org.jgroups.stack.Protocol;
import org.jgroups.util.*;

import java.io.DataInput;
import java.io.DataOutput;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static org.jgroups.Message.Flag.INTERNAL;
import static org.jgroups.Message.Flag.OOB;


/**
 * Group membership protocol. Handles joins/leaves/crashes (suspicions) and
 * emits new views accordingly. Use VIEW_ENFORCER on top of this layer to make
 * sure new members don't receive any messages until they are members
 *
 * @author <NAME>
 */
@MBean(description="Group membership protocol")
public class GMS extends Protocol implements DiagnosticsHandler.ProbeHandler {
    protected static final String CLIENT="Client";
    protected static final String COORD="Coordinator";
    protected static final String PART="Participant";

    // flags for marshalling
    public static final short VIEW_PRESENT     = 1 << 0;
    public static final short DIGEST_PRESENT   = 1 << 1;
    public static final short MERGE_VIEW       = 1 << 2; // if a view is present, is it a MergeView ?
    public static final short DELTA_VIEW       = 1 << 3; // if a view is present, is it a DeltaView ?
    public static final short READ_ADDRS       = 1 << 4; // if digest needs to read its own addresses (rather than that of view)

    /* ------------------------------------------ Properties  ------------------------------------------ */

    @Property(description="Join timeout")
    protected long join_timeout=3000;

    @Property(description="Leave timeout")
    protected long leave_timeout=3000;

    @Property(description="Timeout (in ms) to complete merge")
    protected long merge_timeout=5000; // time to wait for all MERGE_RSPS

    @Property(description="Number of join attempts before we give up and become a singleton. 0 means 'never give up'.")
    protected long max_join_attempts=10;

    @Property(description="Print local address of this member after connect. Default is true")
    protected boolean print_local_addr=true;

    @Property(description="Print physical address(es) on startup")
    protected boolean print_physical_addrs=true;

    /**
     * Whether view bundling (http://jira.jboss.com/jira/browse/JGRP-144) should be enabled or not. Setting this to
     * false forces each JOIN/LEAVE/SUPSECT request to be handled separately. By default these requests are processed
     * together if they are queued at approximately the same time
     */
    @Deprecated
    @Property(description="View bundling toggle",deprecatedMessage="view bundling is enabled by default")
    protected boolean view_bundling=true;

    @Property(description="If true, then GMS is allowed to send VIEW messages with delta views, otherwise " +
      "it always sends full views. See https://issues.jboss.org/browse/JGRP-1354 for details.")
    protected boolean use_delta_views=true;

    @Deprecated
    @Property(description="Max view bundling timeout if view bundling is turned on",
      deprecatedMessage="ignored as view bundling is always on")
    protected long max_bundling_time=50; // 50 ms max to wait for other JOIN, LEAVE or SUSPECT requests

    @Property(description="Max number of old members to keep in history. Default is 50")
    protected int num_prev_mbrs=50;

    @Property(description="Number of views to store in history")
    protected int num_prev_views=10;

    @Property(description="Time in ms to wait for all VIEW acks (0 == wait forever. Default is 2000 msec" )
    protected long view_ack_collection_timeout=2000;

    @Property(description="Use flush for view changes. Default is true")
    protected boolean use_flush_if_present=true;

    @Property(description="Logs failures for collecting all view acks if true")
    protected boolean log_collect_msgs;

    @Property(description="Logs warnings for reception of views less than the current, and for views which don't include self")
    protected boolean log_view_warnings=true;

    /** @deprecated true by default */
    @Property(description="Whether or not to install a new view locally first before broadcasting it " +
      "(only done at the coord ). Set to true automatically if a state transfer protocol is detected",
              deprecatedMessage = "ignored and enabled by default")
    @Deprecated
    protected boolean             install_view_locally_first=true; // https://issues.jboss.org/browse/JGRP-1751

    @ManagedAttribute(description="If true, the current member is in the process of leaving")
    protected volatile boolean    is_leaving;

    /* --------------------------------------------- JMX  ---------------------------------------------- */


    protected int                 num_views;

    /** Stores the last 20 views */
    protected BoundedList<String> prev_views;


    /* --------------------------------------------- Fields ------------------------------------------------ */

    @Property(converter=PropertyConverters.FlushInvoker.class,name="flush_invoker_class")
    protected Class<Callable<Boolean>>  flushInvokerClass;

    protected GmsImpl                   impl;
    protected final Object              impl_mutex=new Object(); // synchronizes event entry into impl
    protected final Map<String,GmsImpl> impls=new HashMap<>(3);

    // Handles merge related tasks
    protected Merger                    merger;

    protected Address                   local_addr;
    protected final Membership          members=new Membership();     // real membership

    protected final Membership          tmp_members=new Membership(); // base for computing next view

    // computes new views and merge views
    protected MembershipChangePolicy    membership_change_policy=new DefaultMembershipPolicy();

    /** Members joined but for which no view has been received yet */
    protected final List<Address>       joining=new ArrayList<>(7);

    /** Members excluded from group, but for which no view has been received yet */
    protected final List<Address>       leaving=new ArrayList<>(7);

    /** To block for the LEAVE-RSP when sending a LEAVE-REQ. The address is the sender of the LEAVE-RSP */
    protected final Promise<Address>    leave_promise=new Promise<>();

    /** Keeps track of old members (up to num_prev_mbrs) */
    protected BoundedList<Address>      prev_members;

    protected volatile View             view;

    protected long                      ltime;

    protected TimeScheduler             timer;

    /** Class to process JOIN, LEAVE and MERGE requests */
    protected final ViewHandler<Request> view_handler=
      new ViewHandler<>(this, this::process, Request::canBeProcessedTogether);

    /** To collect VIEW_ACKs from all members */
    protected final AckCollector        ack_collector=new AckCollector();

    //[JGRP-700] - FLUSH: flushing should span merge
    protected final AckCollector        merge_ack_collector=new AckCollector();

    protected boolean                   flushProtocolInStack;

    // Has this coord sent its first view since becoming coord ? Used to send a full- or delta- view */
    protected boolean                   first_view_sent;



    public GMS() {
    }

    public ViewId getViewId() {return view != null? view.getViewId() : null;}
    public View   view()      {return view;}

    /** Returns the current view and digest. Try to find a matching digest twice (if not found on the first try) */
    public Tuple<View,Digest> getViewAndDigest() {
        MutableDigest digest=new MutableDigest(view.getMembersRaw()).set(getDigest());
        return digest.allSet() || digest.set(getDigest()).allSet()? new Tuple<>(view, digest) : null;
    }

    @ManagedAttribute
    public String getView() {return view != null? view.getViewId().toString() : "null";}
    @ManagedAttribute
    public int getNumberOfViews() {return num_views;}
    @ManagedAttribute
    public String getLocalAddress() {return local_addr != null? local_addr.toString() : "null";}
    @ManagedAttribute
    public String getMembers() {return members.toString();}
    @ManagedAttribute
    public int getNumMembers() {return members.size();}
    public long getJoinTimeout() {return join_timeout;}
    public void setJoinTimeout(long t) {join_timeout=t;}
    public GMS joinTimeout(long timeout) {this.join_timeout=timeout; return this;}
    public GMS leaveTimeout(long timeout) {this.leave_timeout=timeout; return this;}
    public long getMergeTimeout() {return merge_timeout;}
    public void setMergeTimeout(long timeout) {merge_timeout=timeout;}
    public long getMaxJoinAttempts() {return max_join_attempts;}
    public GMS setMaxJoinAttempts(long t) {max_join_attempts=t; return this;}

    @ManagedAttribute(description="impl")
    public String getImplementation() {
        return impl == null? "null" : impl.getClass().getSimpleName();
    }

    @ManagedAttribute(description="Whether or not the current instance is the coordinator")
    public boolean isCoord() {
        return impl instanceof CoordGmsImpl;
    }
<commit_msg>JGRP-2333 pbcast.GMS#getView returns ViewId as opposed to the View<commit_after>package org.jgroups.protocols.pbcast;


import org.jgroups.*;
import org.jgroups.annotations.MBean;
import org.jgroups.annotations.ManagedAttribute;
import org.jgroups.annotations.ManagedOperation;
import org.jgroups.annotations.Property;
import org.jgroups.conf.PropertyConverters;
import org.jgroups.protocols.TP;
import org.jgroups.protocols.pbcast.GmsImpl.Request;
import org.jgroups.stack.DiagnosticsHandler;
import org.jgroups.stack.MembershipChangePolicy;
import org.jgroups.stack.Protocol;
import org.jgroups.util.*;

import java.io.DataInput;
import java.io.DataOutput;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static org.jgroups.Message.Flag.INTERNAL;
import static org.jgroups.Message.Flag.OOB;


/**
 * Group membership protocol. Handles joins/leaves/crashes (suspicions) and
 * emits new views accordingly. Use VIEW_ENFORCER on top of this layer to make
 * sure new members don't receive any messages until they are members
 *
 * @author <NAME>
 */
@MBean(description="Group membership protocol")
public class GMS extends Protocol implements DiagnosticsHandler.ProbeHandler {
    protected static final String CLIENT="Client";
    protected static final String COORD="Coordinator";
    protected static final String PART="Participant";

    // flags for marshalling
    public static final short VIEW_PRESENT     = 1 << 0;
    public static final short DIGEST_PRESENT   = 1 << 1;
    public static final short MERGE_VIEW       = 1 << 2; // if a view is present, is it a MergeView ?
    public static final short DELTA_VIEW       = 1 << 3; // if a view is present, is it a DeltaView ?
    public static final short READ_ADDRS       = 1 << 4; // if digest needs to read its own addresses (rather than that of view)

    /* ------------------------------------------ Properties  ------------------------------------------ */

    @Property(description="Join timeout")
    protected long join_timeout=3000;

    @Property(description="Leave timeout")
    protected long leave_timeout=3000;

    @Property(description="Timeout (in ms) to complete merge")
    protected long merge_timeout=5000; // time to wait for all MERGE_RSPS

    @Property(description="Number of join attempts before we give up and become a singleton. 0 means 'never give up'.")
    protected long max_join_attempts=10;

    @Property(description="Print local address of this member after connect. Default is true")
    protected boolean print_local_addr=true;

    @Property(description="Print physical address(es) on startup")
    protected boolean print_physical_addrs=true;

    /**
     * Whether view bundling (http://jira.jboss.com/jira/browse/JGRP-144) should be enabled or not. Setting this to
     * false forces each JOIN/LEAVE/SUPSECT request to be handled separately. By default these requests are processed
     * together if they are queued at approximately the same time
     */
    @Deprecated
    @Property(description="View bundling toggle",deprecatedMessage="view bundling is enabled by default")
    protected boolean view_bundling=true;

    @Property(description="If true, then GMS is allowed to send VIEW messages with delta views, otherwise " +
      "it always sends full views. See https://issues.jboss.org/browse/JGRP-1354 for details.")
    protected boolean use_delta_views=true;

    @Deprecated
    @Property(description="Max view bundling timeout if view bundling is turned on",
      deprecatedMessage="ignored as view bundling is always on")
    protected long max_bundling_time=50; // 50 ms max to wait for other JOIN, LEAVE or SUSPECT requests

    @Property(description="Max number of old members to keep in history. Default is 50")
    protected int num_prev_mbrs=50;

    @Property(description="Number of views to store in history")
    protected int num_prev_views=10;

    @Property(description="Time in ms to wait for all VIEW acks (0 == wait forever. Default is 2000 msec" )
    protected long view_ack_collection_timeout=2000;

    @Property(description="Use flush for view changes. Default is true")
    protected boolean use_flush_if_present=true;

    @Property(description="Logs failures for collecting all view acks if true")
    protected boolean log_collect_msgs;

    @Property(description="Logs warnings for reception of views less than the current, and for views which don't include self")
    protected boolean log_view_warnings=true;

    /** @deprecated true by default */
    @Property(description="Whether or not to install a new view locally first before broadcasting it " +
      "(only done at the coord ). Set to true automatically if a state transfer protocol is detected",
              deprecatedMessage = "ignored and enabled by default")
    @Deprecated
    protected boolean             install_view_locally_first=true; // https://issues.jboss.org/browse/JGRP-1751

    @ManagedAttribute(description="If true, the current member is in the process of leaving")
    protected volatile boolean    is_leaving;

    /* --------------------------------------------- JMX  ---------------------------------------------- */


    protected int                 num_views;

    /** Stores the last 20 views */
    protected BoundedList<String> prev_views;


    /* --------------------------------------------- Fields ------------------------------------------------ */

    @Property(converter=PropertyConverters.FlushInvoker.class,name="flush_invoker_class")
    protected Class<Callable<Boolean>>  flushInvokerClass;

    protected GmsImpl                   impl;
    protected final Object              impl_mutex=new Object(); // synchronizes event entry into impl
    protected final Map<String,GmsImpl> impls=new HashMap<>(3);

    // Handles merge related tasks
    protected Merger                    merger;

    protected Address                   local_addr;
    protected final Membership          members=new Membership();     // real membership

    protected final Membership          tmp_members=new Membership(); // base for computing next view

    // computes new views and merge views
    protected MembershipChangePolicy    membership_change_policy=new DefaultMembershipPolicy();

    /** Members joined but for which no view has been received yet */
    protected final List<Address>       joining=new ArrayList<>(7);

    /** Members excluded from group, but for which no view has been received yet */
    protected final List<Address>       leaving=new ArrayList<>(7);

    /** To block for the LEAVE-RSP when sending a LEAVE-REQ. The address is the sender of the LEAVE-RSP */
    protected final Promise<Address>    leave_promise=new Promise<>();

    /** Keeps track of old members (up to num_prev_mbrs) */
    protected BoundedList<Address>      prev_members;

    protected volatile View             view;

    protected long                      ltime;

    protected TimeScheduler             timer;

    /** Class to process JOIN, LEAVE and MERGE requests */
    protected final ViewHandler<Request> view_handler=
      new ViewHandler<>(this, this::process, Request::canBeProcessedTogether);

    /** To collect VIEW_ACKs from all members */
    protected final AckCollector        ack_collector=new AckCollector();

    //[JGRP-700] - FLUSH: flushing should span merge
    protected final AckCollector        merge_ack_collector=new AckCollector();

    protected boolean                   flushProtocolInStack;

    // Has this coord sent its first view since becoming coord ? Used to send a full- or delta- view */
    protected boolean                   first_view_sent;



    public GMS() {
    }

    public ViewId getViewId() {return view != null? view.getViewId() : null;}
    public View   view()      {return view;}

    /** Returns the current view and digest. Try to find a matching digest twice (if not found on the first try) */
    public Tuple<View,Digest> getViewAndDigest() {
        MutableDigest digest=new MutableDigest(view.getMembersRaw()).set(getDigest());
        return digest.allSet() || digest.set(getDigest()).allSet()? new Tuple<>(view, digest) : null;
    }

    @ManagedAttribute
    public String getView() {return view != null? view.toString() : "null";}
    @ManagedAttribute
    public int getNumberOfViews() {return num_views;}
    @ManagedAttribute
    public String getLocalAddress() {return local_addr != null? local_addr.toString() : "null";}
    @ManagedAttribute
    public String getMembers() {return members.toString();}
    @ManagedAttribute
    public int getNumMembers() {return members.size();}
    public long getJoinTimeout() {return join_timeout;}
    public void setJoinTimeout(long t) {join_timeout=t;}
    public GMS joinTimeout(long timeout) {this.join_timeout=timeout; return this;}
    public GMS leaveTimeout(long timeout) {this.leave_timeout=timeout; return this;}
    public long getMergeTimeout() {return merge_timeout;}
    public void setMergeTimeout(long timeout) {merge_timeout=timeout;}
    public long getMaxJoinAttempts() {return max_join_attempts;}
    public GMS setMaxJoinAttempts(long t) {max_join_attempts=t; return this;}

    @ManagedAttribute(description="impl")
    public String getImplementation() {
        return impl == null? "null" : impl.getClass().getSimpleName();
    }

    @ManagedAttribute(description="Whether or not the current instance is the coordinator")
    public boolean isCoord() {
        return impl instanceof CoordGmsImpl;
    }

{'fraction_non_alphanumeric': 0.11013376359756415, 'fraction_numerical': 0.006766252016863582, 'mean_word_length': 3.151685393258427, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 37, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '400682', 'n_tokens_mistral': 5164, 'n_tokens_neox': 4865, 'n_words': 1730}