<commit_before>package org.jgroups.protocols;


import org.jgroups.*;
import org.jgroups.Channel;
import org.jgroups.annotations.GuardedBy;
import org.jgroups.stack.IpAddress;
import org.jgroups.stack.Protocol;
import org.jgroups.util.*;
import org.jgroups.util.Queue;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.DataOutputStream;
import java.net.*;
import java.text.NumberFormat;
import java.util.*;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;


/**
 * Generic transport - specific implementations should extend this abstract class.
 * Features which are provided to the subclasses include
 * <ul>
 * <li>version checking
 * <li>marshalling and unmarshalling
 * <li>message bundling (handling single messages, and message lists)
 * <li>incoming packet handler
 * <li>loopback
 * </ul>
 * A subclass has to override
 * <ul>
 * <li>{@link #sendToAllMembers(byte[], int, int)}
 * <li>{@link #sendToSingleMember(org.jgroups.Address, byte[], int, int)}
 * <li>{@link #init()}
 * <li>{@link #start()}: subclasses <em>must</em> call super.start() <em>after</em> they initialize themselves
 * (e.g., created their sockets).
 * <li>{@link #stop()}: subclasses <em>must</em> call super.stop() after they deinitialized themselves
 * <li>{@link #destroy()}
 * </ul>
 * The create() or start() method has to create a local address.<br>
 * The {@link #receive(Address, Address, byte[], int, int)} method must
 * be called by subclasses when a unicast or multicast message has been received.
 * @author <NAME>
 * @version $Id: TP.java,v 1.146 2007/06/04 13:37:20 belaban Exp $
 */
public abstract class TP extends Protocol {


    /** The address (host and port) of this member */
    protected Address         local_addr=null;

    /** The name of the group to which this member is connected */
    protected String          channel_name=null;

    /** The interface (NIC) which should be used by this transport */
    protected InetAddress     bind_addr=null;

    /** Overrides bind_addr, -Djgroups.bind_addr and -Dbind.address: let's the OS return the local host address */
    boolean         use_local_host=false;

    /** If true, the transport should use all available interfaces to receive multicast messages */
    boolean         receive_on_all_interfaces=false;

    /** List<NetworkInterface> of interfaces to receive multicasts on. The multicast receive socket will listen
     * on all of these interfaces. This is a comma-separated list of IP addresses or interface names. E.g.
     * "192.168.5.1,eth1,127.0.0.1". Duplicates are discarded; we only bind to an interface once.
     * If this property is set, it override receive_on_all_interfaces.
     */
    java.util.List  receive_interfaces=null;

    /** If true, the transport should use all available interfaces to send multicast messages. This means
     * the same multicast message is sent N times, so use with care */
    boolean         send_on_all_interfaces=false;

    /** List<NetworkInterface> of interfaces to send multicasts on. The multicast send socket will send the
     * same multicast message on all of these interfaces. This is a comma-separated list of IP addresses or
     * interface names. E.g. "192.168.5.1,eth1,127.0.0.1". Duplicates are discarded.
     * If this property is set, it override send_on_all_interfaces.
     */
    java.util.List  send_interfaces=null;


    /** The port to which the transport binds. 0 means to bind to any (ephemeral) port */
    int             bind_port=0;
    int				port_range=1; // 27-6-2003 bgooren, Only try one port by default

    /** The members of this group (updated when a member joins or leaves) */
    final protected Vector<Address> members=new Vector<Address>(11);

    protected View            view=null;


    final ExposedByteArrayInputStream  in_stream=new ExposedByteArrayInputStream(new byte[]{'0'});
    final DataInputStream              dis=new DataInputStream(in_stream);


    /** If true, messages sent to self are treated specially: unicast messages are
     * looped back immediately, multicast messages get a local copy first and -
     * when the real copy arrives - it will be discarded. Useful for Window
     * media (non)sense */
    boolean         loopback=false;


    /** Discard packets with a different version. Usually minor version differences are okay. Setting this property
     * to true means that we expect the exact same version on all incoming packets */
    protected boolean discard_incompatible_packets=false;

    /** Sometimes receivers are overloaded (they have to handle de-serialization etc).
     * Packet handler is a separate thread taking care of de-serialization, receiver
     * thread(s) simply put packet in queue and return immediately. Setting this to
     * true adds one more thread */
    boolean         use_incoming_packet_handler=true;

    /** Used by packet handler to store incoming DatagramPackets */
    Queue           incoming_packet_queue=null;

    /** Dequeues DatagramPackets from packet_queue, unmarshalls them and
     * calls <tt>handleIncomingUdpPacket()</tt> */
    IncomingPacketHandler   incoming_packet_handler=null;


    /** Used by packet handler to store incoming Messages */
    Queue                  incoming_msg_queue=null;

    IncomingMessageHandler incoming_msg_handler;


    boolean use_concurrent_stack=true;
    ThreadGroup pool_thread_group=new ThreadGroup(Util.getGlobalThreadGroup(), "Thread Pools");

    /**
     * Names the current thread. Valid values are "pcl":
     * p: include the previous (original) name, e.g. "Incoming thread-1", "UDP ucast receiver"
     * c: include the cluster name, e.g. "MyCluster"
     * l: include the local address of the current member, e.g. "192.168.5.1:5678"
     */
    protected ThreadNamingPattern thread_naming_pattern;


    /** ================================== OOB thread pool ============================== */
    /** The thread pool which handles OOB messages */
    Executor oob_thread_pool;
    boolean oob_thread_pool_enabled=true;
    int oob_thread_pool_min_threads=2;
    int oob_thread_pool_max_threads=10;
    /** Number of milliseconds after which an idle thread is removed */
    long oob_thread_pool_keep_alive_time=30000;

    long num_oob_msgs_received=0;

    /** Used if oob_thread_pool is a ThreadPoolExecutor and oob_thread_pool_queue_enabled is true */
    BlockingQueue<Runnable> oob_thread_pool_queue=null;
    /** Whether of not to use a queue with ThreadPoolExecutor (ignored with direct executor) */
    boolean oob_thread_pool_queue_enabled=true;
    /** max number of elements in queue (bounded) */
    int oob_thread_pool_queue_max_size=500;
    /** Possible values are "Abort", "Discard", "DiscardOldest" and "Run". These values might change once we switch to
     * JDK 5's java.util.concurrent package */
    String oob_thread_pool_rejection_policy="Run";


/** ================================== Regular thread pool ============================== */

    /** The thread pool which handles unmarshalling, version checks and dispatching of regular messages */
    Executor thread_pool;<commit_msg>defaulted thread_naming_pattern to "cl"<commit_after>package org.jgroups.protocols;


import org.jgroups.*;
import org.jgroups.Channel;
import org.jgroups.annotations.GuardedBy;
import org.jgroups.stack.IpAddress;
import org.jgroups.stack.Protocol;
import org.jgroups.util.*;
import org.jgroups.util.Queue;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.DataOutputStream;
import java.net.*;
import java.text.NumberFormat;
import java.util.*;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;


/**
 * Generic transport - specific implementations should extend this abstract class.
 * Features which are provided to the subclasses include
 * <ul>
 * <li>version checking
 * <li>marshalling and unmarshalling
 * <li>message bundling (handling single messages, and message lists)
 * <li>incoming packet handler
 * <li>loopback
 * </ul>
 * A subclass has to override
 * <ul>
 * <li>{@link #sendToAllMembers(byte[], int, int)}
 * <li>{@link #sendToSingleMember(org.jgroups.Address, byte[], int, int)}
 * <li>{@link #init()}
 * <li>{@link #start()}: subclasses <em>must</em> call super.start() <em>after</em> they initialize themselves
 * (e.g., created their sockets).
 * <li>{@link #stop()}: subclasses <em>must</em> call super.stop() after they deinitialized themselves
 * <li>{@link #destroy()}
 * </ul>
 * The create() or start() method has to create a local address.<br>
 * The {@link #receive(Address, Address, byte[], int, int)} method must
 * be called by subclasses when a unicast or multicast message has been received.
 * @author <NAME>
 * @version $Id: TP.java,v 1.147 2007/06/06 06:39:25 belaban Exp $
 */
public abstract class TP extends Protocol {


    /** The address (host and port) of this member */
    protected Address         local_addr=null;

    /** The name of the group to which this member is connected */
    protected String          channel_name=null;

    /** The interface (NIC) which should be used by this transport */
    protected InetAddress     bind_addr=null;

    /** Overrides bind_addr, -Djgroups.bind_addr and -Dbind.address: let's the OS return the local host address */
    boolean         use_local_host=false;

    /** If true, the transport should use all available interfaces to receive multicast messages */
    boolean         receive_on_all_interfaces=false;

    /** List<NetworkInterface> of interfaces to receive multicasts on. The multicast receive socket will listen
     * on all of these interfaces. This is a comma-separated list of IP addresses or interface names. E.g.
     * "192.168.5.1,eth1,127.0.0.1". Duplicates are discarded; we only bind to an interface once.
     * If this property is set, it override receive_on_all_interfaces.
     */
    java.util.List  receive_interfaces=null;

    /** If true, the transport should use all available interfaces to send multicast messages. This means
     * the same multicast message is sent N times, so use with care */
    boolean         send_on_all_interfaces=false;

    /** List<NetworkInterface> of interfaces to send multicasts on. The multicast send socket will send the
     * same multicast message on all of these interfaces. This is a comma-separated list of IP addresses or
     * interface names. E.g. "192.168.5.1,eth1,127.0.0.1". Duplicates are discarded.
     * If this property is set, it override send_on_all_interfaces.
     */
    java.util.List  send_interfaces=null;


    /** The port to which the transport binds. 0 means to bind to any (ephemeral) port */
    int             bind_port=0;
    int				port_range=1; // 27-6-2003 bgooren, Only try one port by default

    /** The members of this group (updated when a member joins or leaves) */
    final protected Vector<Address> members=new Vector<Address>(11);

    protected View            view=null;


    final ExposedByteArrayInputStream  in_stream=new ExposedByteArrayInputStream(new byte[]{'0'});
    final DataInputStream              dis=new DataInputStream(in_stream);


    /** If true, messages sent to self are treated specially: unicast messages are
     * looped back immediately, multicast messages get a local copy first and -
     * when the real copy arrives - it will be discarded. Useful for Window
     * media (non)sense */
    boolean         loopback=false;


    /** Discard packets with a different version. Usually minor version differences are okay. Setting this property
     * to true means that we expect the exact same version on all incoming packets */
    protected boolean discard_incompatible_packets=false;

    /** Sometimes receivers are overloaded (they have to handle de-serialization etc).
     * Packet handler is a separate thread taking care of de-serialization, receiver
     * thread(s) simply put packet in queue and return immediately. Setting this to
     * true adds one more thread */
    boolean         use_incoming_packet_handler=true;

    /** Used by packet handler to store incoming DatagramPackets */
    Queue           incoming_packet_queue=null;

    /** Dequeues DatagramPackets from packet_queue, unmarshalls them and
     * calls <tt>handleIncomingUdpPacket()</tt> */
    IncomingPacketHandler   incoming_packet_handler=null;


    /** Used by packet handler to store incoming Messages */
    Queue                  incoming_msg_queue=null;

    IncomingMessageHandler incoming_msg_handler;


    boolean use_concurrent_stack=true;
    ThreadGroup pool_thread_group=new ThreadGroup(Util.getGlobalThreadGroup(), "Thread Pools");

    /**
     * Names the current thread. Valid values are "pcl":
     * p: include the previous (original) name, e.g. "Incoming thread-1", "UDP ucast receiver"
     * c: include the cluster name, e.g. "MyCluster"
     * l: include the local address of the current member, e.g. "192.168.5.1:5678"
     */
    protected ThreadNamingPattern thread_naming_pattern=new ThreadNamingPattern("cl");


    /** ================================== OOB thread pool ============================== */
    /** The thread pool which handles OOB messages */
    Executor oob_thread_pool;
    boolean oob_thread_pool_enabled=true;
    int oob_thread_pool_min_threads=2;
    int oob_thread_pool_max_threads=10;
    /** Number of milliseconds after which an idle thread is removed */
    long oob_thread_pool_keep_alive_time=30000;

    long num_oob_msgs_received=0;

    /** Used if oob_thread_pool is a ThreadPoolExecutor and oob_thread_pool_queue_enabled is true */
    BlockingQueue<Runnable> oob_thread_pool_queue=null;
    /** Whether of not to use a queue with ThreadPoolExecutor (ignored with direct executor) */
    boolean oob_thread_pool_queue_enabled=true;
    /** max number of elements in queue (bounded) */
    int oob_thread_pool_queue_max_size=500;
    /** Possible values are "Abort", "Discard", "DiscardOldest" and "Run". These values might change once we switch to
     * JDK 5's java.util.concurrent package */
    String oob_thread_pool_rejection_policy="Run";


/** ================================== Regular thread pool ============================== */

    /** The thread pool which handles unmarshalling, version checks and dispatching of regular messages */
    Executor thread_pool;
{'fraction_non_alphanumeric': 0.10681157413816973, 'fraction_numerical': 0.012044856707739167, 'mean_word_length': 3.6260006404098624, 'pattern_counts': {'":': 2, '<': 63, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 10, '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': '4894648', 'n_tokens_mistral': 4212, 'n_tokens_neox': 3946, 'n_words': 1433}