Package org.jgroups.protocols

Class TP

java.lang.Object
org.jgroups.stack.Protocol
org.jgroups.protocols.TP
Direct Known Subclasses:
BasicTCP, LOOPBACK, SHARED_LOOPBACK, TUNNEL, UDP
public abstract class TP extends Protocol
Generic transport - specific implementations should extend this abstract class. Features which are provided to the subclasses include
  • version checking
  • marshalling and unmarshalling
  • message bundling (handling single messages, and message lists)
  • incoming packet handler
  • loopback
A subclass has to override The create() or start() method has to create a local address.
The receive(Address, byte[], int, int) method must be called by subclasses when a unicast or multicast message has been received.
Author:
Bela Ban

  • Field Details

    • LIST

      protected static final byte LIST
      See Also:

    • MULTICAST

      protected static final byte MULTICAST
      See Also:

    • OOB

      protected static final byte OOB
      See Also:

    • can_bind_to_mcast_addr

      protected static final boolean can_bind_to_mcast_addr

    • f

      protected static NumberFormat f

    • bind_addr

      protected InetAddress bind_addr

    • bind_interface_str

      protected String bind_interface_str

    • receive_on_all_interfaces

      protected boolean receive_on_all_interfaces

    • receive_interfaces

      protected List<NetworkInterface> receive_interfaces
      List 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 overrides receive_on_all_interfaces.

    • logical_addr_cache_max_size

      protected int logical_addr_cache_max_size

    • logical_addr_cache_expiration

      protected long logical_addr_cache_expiration

    • bind_port

      protected int bind_port
      The port to which the transport binds. 0 means to bind to any (ephemeral) port

    • port_range

      protected int port_range

    • loopback

      protected boolean loopback
      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

    • discard_incompatible_packets

      protected boolean discard_incompatible_packets
      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

    • thread_naming_pattern

      protected String thread_naming_pattern

    • oob_thread_pool_enabled

      protected boolean oob_thread_pool_enabled

    • oob_thread_pool_min_threads

      protected int oob_thread_pool_min_threads

    • oob_thread_pool_max_threads

      protected int oob_thread_pool_max_threads

    • oob_thread_pool_keep_alive_time

      protected long oob_thread_pool_keep_alive_time

    • oob_thread_pool_queue_enabled

      protected boolean oob_thread_pool_queue_enabled

    • oob_thread_pool_queue_max_size

      protected int oob_thread_pool_queue_max_size

    • thread_pool_min_threads

      protected int thread_pool_min_threads

    • thread_pool_max_threads

      protected int thread_pool_max_threads

    • thread_pool_keep_alive_time

      protected long thread_pool_keep_alive_time

    • thread_pool_enabled

      protected boolean thread_pool_enabled

    • thread_pool_queue_enabled

      protected boolean thread_pool_queue_enabled

    • thread_pool_queue_max_size

      protected int thread_pool_queue_max_size

    • thread_pool_rejection_policy

      protected String thread_pool_rejection_policy

    • timer_type

      protected String timer_type

    • timer_min_threads

      protected int timer_min_threads

    • timer_max_threads

      protected int timer_max_threads

    • timer_keep_alive_time

      protected long timer_keep_alive_time

    • timer_queue_max_size

      protected int timer_queue_max_size

    • wheel_size

      protected int wheel_size

    • tick_time

      protected long tick_time

    • enable_bundling

      protected boolean enable_bundling

    • enable_unicast_bundling

      protected boolean enable_unicast_bundling
      Enable bundling for unicast messages. Ignored if enable_bundling is off

    • enable_diagnostics

      protected boolean enable_diagnostics

    • diagnostics_addr

      protected InetAddress diagnostics_addr

    • diagnostics_port

      protected int diagnostics_port

    • singleton_name

      protected String singleton_name

    • log_discard_msgs

      protected boolean log_discard_msgs
      Whether or not warnings about messages from different groups are logged - private flag, not for common use

    • max_bundle_size

      protected int max_bundle_size
      Maximum number of bytes for messages to be queued until they are sent. This value needs to be smaller than the largest datagram packet size in case of UDP

    • max_bundle_timeout

      protected long max_bundle_timeout
      Max number of milliseconds until queued messages are sent. Messages are sent when max_bundle_size or max_bundle_timeout has been exceeded (whichever occurs faster)

    • bundler_type

      protected String bundler_type

    • bundler_capacity

      protected int bundler_capacity

    • num_msgs_sent

      protected long num_msgs_sent

    • num_msgs_received

      protected long num_msgs_received

    • num_bytes_sent

      protected long num_bytes_sent

    • num_bytes_received

      protected long num_bytes_received

    • channel_name

      protected String channel_name
      The name of the group to which this member is connected. With a shared transport, the channel name is in TP.ProtocolAdapter (cluster_name), and this field is not used

    • num_oob_msgs_received

      protected long num_oob_msgs_received

    • num_incoming_msgs_received

      protected long num_incoming_msgs_received

    • local_addr

      protected Address local_addr
      The address (host and port) of this member. Null by default when a shared transport is used

    • members

      protected final Set<Address> members
      The members of this group (updated when a member joins or leaves). With a shared transport, members contains *all* members from all channels sitting on the shared transport

    • pool_thread_group

      protected ThreadGroup pool_thread_group

    • connect_count

      protected int connect_count
      Keeps track of connects and disconnects, in order to start and stop threads

    • connectLock

      protected final ReentrantLock connectLock

    • oob_thread_pool

      protected Executor oob_thread_pool

    • oob_thread_factory

      protected ThreadFactory oob_thread_factory
      Factory which is used by oob_thread_pool

    • oob_thread_pool_queue

      protected BlockingQueue<Runnable> oob_thread_pool_queue
      Used if oob_thread_pool is a ThreadPoolExecutor and oob_thread_pool_queue_enabled is true

    • thread_pool

      protected Executor thread_pool
      The thread pool which handles unmarshalling, version checks and dispatching of regular messages

    • default_thread_factory

      protected ThreadFactory default_thread_factory
      Factory which is used by oob_thread_pool

    • thread_pool_queue

      protected BlockingQueue<Runnable> thread_pool_queue
      Used if thread_pool is a ThreadPoolExecutor and thread_pool_queue_enabled is true

    • timer

      protected TimeScheduler timer

    • timer_thread_factory

      protected ThreadFactory timer_thread_factory

    • global_thread_factory

      protected ThreadFactory global_thread_factory
      Used by all threads created by JGroups outside of the thread pools

    • socket_factory

      protected SocketFactory socket_factory

    • bundler

      protected TP.Bundler bundler

    • diag_handler

      protected TP.DiagnosticsHandler diag_handler

    • preregistered_probe_handlers

      protected final List<TP.ProbeHandler> preregistered_probe_handlers

    • up_prots

      protected final ConcurrentMap<String,Protocol> up_prots
      If singleton_name is enabled, this map is used to de-multiplex incoming messages according to their cluster names (attached to the message by the transport anyway). The values are the next protocols above the transports.

    • logical_addr_cache

      protected LazyRemovalCache<Address,PhysicalAddress> logical_addr_cache
      Cache which maintains mappings between logical and physical addresses. When sending a message to a logical address, we look up the physical address from logical_addr_cache and send the message to the physical address
      The keys are logical addresses, the values physical addresses

    • last_discovery_request

      protected long last_discovery_request

    • who_has_cache

      protected AgeOutCache<Address> who_has_cache
      Cache keeping track of WHO_HAS requests for physical addresses (given a logical address) and expiring them after 5000ms

  • Constructor Details

    • TP

      protected TP()
      Creates the TP protocol, and initializes the state variables, does however not start any sockets or threads.

  • Method Details

    • setMaxBundleSize

      public void setMaxBundleSize(int size)

    • getMaxBundleTimeout

      public long getMaxBundleTimeout()

    • setMaxBundleTimeout

      public void setMaxBundleTimeout(long timeout)

    • getMaxBundleSize

      public int getMaxBundleSize()

    • getBundlerBufferSize

      public int getBundlerBufferSize()

    • setOOBThreadPoolKeepAliveTime

      public void setOOBThreadPoolKeepAliveTime(long time)

    • getOOBThreadPoolKeepAliveTime

      public long getOOBThreadPoolKeepAliveTime()

    • setOOBThreadPoolMinThreads

      public void setOOBThreadPoolMinThreads(int size)

    • getOOBThreadPoolMinThreads

      public int getOOBThreadPoolMinThreads()

    • setOOBThreadPoolMaxThreads

      public void setOOBThreadPoolMaxThreads(int size)

    • getOOBThreadPoolMaxThreads

      public int getOOBThreadPoolMaxThreads()

    • setThreadPoolMinThreads

      public void setThreadPoolMinThreads(int size)

    • getThreadPoolMinThreads

      public int getThreadPoolMinThreads()

    • setThreadPoolMaxThreads

      public void setThreadPoolMaxThreads(int size)

    • getThreadPoolMaxThreads

      public int getThreadPoolMaxThreads()

    • setThreadPoolKeepAliveTime

      public void setThreadPoolKeepAliveTime(long time)

    • getThreadPoolKeepAliveTime

      public long getThreadPoolKeepAliveTime()

    • setTimerMinThreads

      public void setTimerMinThreads(int size)

    • getTimerMinThreads

      public int getTimerMinThreads()

    • setTimerMaxThreads

      public void setTimerMaxThreads(int size)

    • getTimerMaxThreads

      public int getTimerMaxThreads()

    • setTimerKeepAliveTime

      public void setTimerKeepAliveTime(long time)

    • getTimerKeepAliveTime

      public long getTimerKeepAliveTime()

    • getTimerQueueSize

      public int getTimerQueueSize()

    • getTimerClass

      public String getTimerClass()

    • supportsMulticasting

      public abstract boolean supportsMulticasting()
      Whether or not hardware multicasting is supported

    • toString

      public String toString()
      Overrides:
      toString in class Object

    • resetStats

      public void resetStats()
      Overrides:
      resetStats in class Protocol

    • registerProbeHandler

      public void registerProbeHandler(TP.ProbeHandler handler)

    • unregisterProbeHandler

      public void unregisterProbeHandler(TP.ProbeHandler handler)

    • getPoolThreadGroup

      public ThreadGroup getPoolThreadGroup()

    • setThreadPoolQueueEnabled

      public void setThreadPoolQueueEnabled(boolean flag)

    • getDefaultThreadPool

      public Executor getDefaultThreadPool()

    • setDefaultThreadPool

      public void setDefaultThreadPool(Executor thread_pool)

    • getDefaultThreadPoolThreadFactory

      public ThreadFactory getDefaultThreadPoolThreadFactory()

    • setDefaultThreadPoolThreadFactory

      public void setDefaultThreadPoolThreadFactory(ThreadFactory factory)

    • getOOBThreadPool

      public Executor getOOBThreadPool()

    • setOOBThreadPool

      public void setOOBThreadPool(Executor oob_thread_pool)

    • getOOBThreadPoolThreadFactory

      public ThreadFactory getOOBThreadPoolThreadFactory()

    • setOOBThreadPoolThreadFactory

      public void setOOBThreadPoolThreadFactory(ThreadFactory factory)

    • getTimerThreadFactory

      public ThreadFactory getTimerThreadFactory()

    • setTimerThreadFactory

      public void setTimerThreadFactory(ThreadFactory factory)

    • getTimer

      public TimeScheduler getTimer()

    • getThreadFactory

      public ThreadFactory getThreadFactory()
      Description copied from class: Protocol
      Supposed to be overwritten by subclasses. Usually the transport returns a valid non-null thread factory, but thread factories can also be created by individual protocols
      Overrides:
      getThreadFactory in class Protocol
      Returns:

    • setThreadFactory

      public void setThreadFactory(ThreadFactory factory)

    • getSocketFactory

      public SocketFactory getSocketFactory()
      Description copied from class: Protocol
      Returns the SocketFactory associated with this protocol, if overridden in a subclass, or passes the call down
      Overrides:
      getSocketFactory in class Protocol
      Returns:
      SocketFactory

    • setSocketFactory

      public void setSocketFactory(SocketFactory factory)
      Description copied from class: Protocol
      Sets a SocketFactory. Socket factories are typically provided by the transport (TP) or TP.ProtocolAdapter
      Overrides:
      setSocketFactory in class Protocol
      Parameters:
      factory -

    • getThreadNamingPattern

      public String getThreadNamingPattern()
      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"

    • getNumMessagesSent

      public long getNumMessagesSent()

    • getNumMessagesReceived

      public long getNumMessagesReceived()

    • getNumBytesSent

      public long getNumBytesSent()

    • getNumBytesReceived

      public long getNumBytesReceived()

    • getBindAddress

      public String getBindAddress()

    • setBindAddress

      public void setBindAddress(String bind_addr) throws UnknownHostException
      Throws:
      UnknownHostException

    • getBindAddressAsInetAddress

      public InetAddress getBindAddressAsInetAddress()

    • getBindPort

      public int getBindPort()

    • setBindPort

      public void setBindPort(int port)

    • getBindToAllInterfaces

      public boolean getBindToAllInterfaces()
      Deprecated.
      Use isReceiveOnAllInterfaces() instead

    • setBindToAllInterfaces

      public void setBindToAllInterfaces(boolean flag)

    • isReceiveOnAllInterfaces

      public boolean isReceiveOnAllInterfaces()

    • getReceiveInterfaces

      public List<NetworkInterface> getReceiveInterfaces()

    • isSendOnAllInterfaces

      public static boolean isSendOnAllInterfaces()
      Deprecated.
      This property was removed in 2.7

    • getSendInterfaces

      public static List<NetworkInterface> getSendInterfaces()
      Deprecated.
      This property was removed in 2.7

    • isDiscardIncompatiblePackets

      public boolean isDiscardIncompatiblePackets()

    • setDiscardIncompatiblePackets

      public void setDiscardIncompatiblePackets(boolean flag)

    • isEnableBundling

      public boolean isEnableBundling()

    • setEnableBundling

      public void setEnableBundling(boolean flag)

    • isEnableUnicastBundling

      public boolean isEnableUnicastBundling()

    • setEnableUnicastBundling

      public void setEnableUnicastBundling(boolean enable_unicast_bundling)

    • setPortRange

      public void setPortRange(int range)

    • getPortRange

      public int getPortRange()

    • setUseConcurrentStack

      @Deprecated public void setUseConcurrentStack(boolean flag)
      Deprecated.
      the concurrent stack is used by default

    • isOOBThreadPoolEnabled

      public boolean isOOBThreadPoolEnabled()

    • isDefaulThreadPoolEnabled

      public boolean isDefaulThreadPoolEnabled()

    • isLoopback

      public boolean isLoopback()

    • setLoopback

      public void setLoopback(boolean b)

    • isUseIncomingPacketHandler

      public static boolean isUseIncomingPacketHandler()
      Deprecated.
      With the concurrent stack being the default, this property is ignored

    • getUpProtocols

      public ConcurrentMap<String,Protocol> getUpProtocols()

    • getOOBPoolSize

      public int getOOBPoolSize()

    • getOOBMessages

      public long getOOBMessages()

    • getOOBQueueSize

      public int getOOBQueueSize()

    • getOOBMaxQueueSize

      public int getOOBMaxQueueSize()

    • setOOBRejectionPolicy

      public void setOOBRejectionPolicy(String rejection_policy)

    • getRegularPoolSize

      public int getRegularPoolSize()

    • getRegularMessages

      public long getRegularMessages()

    • getRegularQueueSize

      public int getRegularQueueSize()

    • getRegularMaxQueueSize

      public int getRegularMaxQueueSize()

    • getNumTimerTasks

      public int getNumTimerTasks()

    • dumpTimerTasks

      public String dumpTimerTasks()

    • getTimerThreads

      public int getTimerThreads()

    • setRegularRejectionPolicy

      public void setRegularRejectionPolicy(String rejection_policy)

    • setLogDiscardMessages

      public void setLogDiscardMessages(boolean flag)

    • getLogDiscardMessages

      public boolean getLogDiscardMessages()

    • printLogicalAddressCache

      public String printLogicalAddressCache()

    • evictLogicalAddressCache

      public void evictLogicalAddressCache()

    • sendMulticast

      public abstract void sendMulticast(byte[] data, int offset, int length) throws Exception
      Send to all members in the group. UDP would use an IP multicast message, whereas TCP would send N messages, one for each member
      Parameters:
      data - The data to be sent. This is not a copy, so don't modify it
      offset -
      length -
      Throws:
      Exception

    • sendUnicast

      public abstract void sendUnicast(PhysicalAddress dest, byte[] data, int offset, int length) throws Exception
      Send a unicast to 1 member. Note that the destination address is a *physical*, not a logical address
      Parameters:
      dest - Must be a non-null unicast address
      data - The data to be sent. This is not a copy, so don't modify it
      offset -
      length -
      Throws:
      Exception

    • getInfo

      public abstract String getInfo()

    • init

      public void init() throws Exception
      Description copied from class: Protocol
      Called after instance has been created (null constructor) and before protocol is started. Properties are already set. Other protocols are not yet connected and events cannot yet be sent.
      Overrides:
      init in class Protocol
      Throws:
      Exception - Thrown if protocol cannot be initialized successfully. This will cause the ProtocolStack to fail, so the channel constructor will throw an exception

    • destroy

      public void destroy()
      Description copied from class: Protocol
      This method is called on a Channel.close(). Does some cleanup; after the call the VM will terminate
      Overrides:
      destroy in class Protocol

    • start

      public void start() throws Exception
      Creates the unicast and multicast sockets and starts the unicast and multicast receiver threads
      Overrides:
      start in class Protocol
      Throws:
      Exception - Thrown if protocol cannot be started successfully. This will cause the ProtocolStack to fail, so Channel.connect(String) will throw an exception

    • stop

      public void stop()
      Description copied from class: Protocol
      This method is called on a Channel.disconnect(). Stops work (e.g. by closing multicast socket). Will be called from top to bottom. This means that at the time of the method invocation the neighbor protocol below is still working. This method will replace the STOP, STOP_OK, CLEANUP and CLEANUP_OK events. The ProtocolStack guarantees that when this method is called all messages in the down queue will have been flushed
      Overrides:
      stop in class Protocol

    • handleConnect

      protected void handleConnect() throws Exception
      Throws:
      Exception

    • handleDisconnect

      protected void handleDisconnect()

    • getSingletonName

      public String getSingletonName()

    • isSingleton

      public boolean isSingleton()

    • up

      public Object up(Event evt)
      handle the UP event.
      Overrides:
      up in class Protocol
      Parameters:
      evt - - the event being send from the stack

    • down

      public Object down(Event evt)
      Caller by the layer above this layer. Usually we just put this Message into the send queue and let one or more worker threads handle it. A worker thread then removes the Message from the send queue, performs a conversion and adds the modified Message to the send queue of the layer below it, by calling down()).
      Overrides:
      down in class Protocol

    • setSourceAddress

      protected void setSourceAddress(Message msg)
      If the sender is null, set our own address. We cannot just go ahead and set the address anyway, as we might be sending a message on behalf of someone else ! E.g. in case of retransmission, when the original sender has crashed, or in a FLUSH protocol when we have to return all unstable messages with the FLUSH_OK response.

    • passMessageUp

      protected void passMessageUp(Message msg, boolean perform_cluster_name_matching, boolean multicast, boolean discard_own_mcast)

    • receive

      protected void receive(Address sender, byte[] data, int offset, int length)
      Subclasses must call this method when a unicast or multicast message has been received. Declared final so subclasses cannot override this method.
      Parameters:
      sender -
      data -
      offset -
      length -

    • dispatchToThreadPool

      protected void dispatchToThreadPool(Executor pool, Address sender, byte[] data, int offset, int length)

    • send

      protected void send(Message msg, Address dest, boolean multicast) throws Exception
      Serializes and sends a message. This method is not reentrant
      Throws:
      Exception

    • doSend

      protected void doSend(Buffer buf, Address dest, boolean multicast) throws Exception
      Throws:
      Exception

    • sendToSingleMember

      protected void sendToSingleMember(Address dest, byte[] buf, int offset, int length) throws Exception
      Throws:
      Exception

    • sendToAllPhysicalAddresses

      protected void sendToAllPhysicalAddresses(byte[] buf, int offset, int length) throws Exception
      Throws:
      Exception

    • writeMessage

      protected static void writeMessage(Message msg, DataOutputStream dos, boolean multicast) throws Exception
      This method needs to be synchronized on out_stream when it is called
      Parameters:
      msg -
      Throws:
      IOException
      Exception

    • readMessage

      protected static Message readMessage(DataInputStream instream) throws Exception
      Throws:
      Exception

    • writeMessageList

      protected static void writeMessageList(Address dest, Address src, List<Message> msgs, DataOutputStream dos, boolean multicast) throws Exception
      Write a lits of messages with the same destination and *mostly* the same src addresses. The message list is marshalled as follows: 
       List: * | version | flags | dest | src | [Message*] |

 Message:  | presence | leading | flags | [src] | length | [buffer] | size | [Headers*] |
      Parameters:
      dest -
      src -
      msgs -
      dos -
      multicast -
      Throws:
      Exception

    • readMessageList

      protected static List<Message> readMessageList(DataInputStream in) throws Exception
      Throws:
      Exception

    • handleDownEvent

      protected Object handleDownEvent(Event evt)

    • registerLocalAddress

      protected void registerLocalAddress(Address addr)
      Associates the address with the physical address fetched from the cache
      Parameters:
      addr -

    • fetchLocalAddresses

      protected void fetchLocalAddresses()
      Grabs the local address (or addresses in the shared transport case) and registers them with the physical address in the transport's cache

    • setThreadNames

      protected void setThreadNames()

    • unsetThreadNames

      protected void unsetThreadNames()

    • setInAllThreadFactories

      protected void setInAllThreadFactories(String cluster_name, Address local_address, String pattern)

    • createThreadPool

      protected static ExecutorService createThreadPool(int min_threads, int max_threads, long keep_alive_time, String rejection_policy, BlockingQueue<Runnable> queue, ThreadFactory factory)

    • shutdownThreadPool

      protected static void shutdownThreadPool(Executor thread_pool)

    • verifyRejectionPolicy

      protected void verifyRejectionPolicy(String str) throws Exception
      Throws:
      Exception

    • parseRejectionPolicy

      protected static RejectedExecutionHandler parseRejectionPolicy(String rejection_policy)

    • passToAllUpProtocols

      protected void passToAllUpProtocols(Event evt)

    • addPhysicalAddressToCache

      protected void addPhysicalAddressToCache(Address logical_addr, PhysicalAddress physical_addr)

    • getPhysicalAddressFromCache

      protected PhysicalAddress getPhysicalAddressFromCache(Address logical_addr)

    • removeLogicalAddressFromCache

      protected void removeLogicalAddressFromCache(Address logical_addr)

    • clearLogicalAddressCache

      public void clearLogicalAddressCache()
      Clears the cache. Do not use, this is only for unit testing !

    • getPhysicalAddress

      protected abstract PhysicalAddress getPhysicalAddress()