Class TP

java.lang.Object

org.jgroups.stack.Protocol

org.jgroups.protocols.TPConfig

org.jgroups.protocols.TP

All Implemented Interfaces:

Lifecycle, DiagnosticsHandler.ProbeHandler

Direct Known Subclasses:

BasicTCP, MockTransport, SHARED_LOOPBACK, TUNNEL, UDP

public abstract class TP
extends TPConfig
implements DiagnosticsHandler.ProbeHandler

Generic transport - specific implementations should extend this 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

A subclass has to override

• sendUnicast(PhysicalAddress, ByteBuffer)
• init()
• start(): subclasses must call super.start() after they initialize themselves (e.g., created their sockets).
• stop(): subclasses must call super.stop() after they deinitialized themselves
• destroy()

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 Summary

Fields

Modifier and Type	Field	Description
static final byte	LIST
protected static final long	MIN_WAIT_BETWEEN_DISCOVERIES
static final int	MSG_OVERHEAD
static final byte	MULTICAST

Fields inherited from class TPConfig

async_executor, bind_addr, bind_port, bundler, bundler_type, cluster_name, connectLock, diag_handler, external_addr, external_port, header, is_trace, last_discovery_request, local_physical_addr, local_transport, local_transport_class, log_discard_msgs, log_discard_msgs_version, logical_addr_cache, logical_addr_cache_expiration, logical_addr_cache_max_size, logical_addr_cache_reaper, logical_addr_cache_reaper_interval, members, message_processing_policy, msg_processing_policy, msg_stats, port_range, print_function, receive_interfaces, receive_on_all_interfaces, rtt, socket_factory, suppress_log_different_cluster, suppress_log_different_version, suppress_time_different_cluster_warnings, suppress_time_different_version_warnings, thread_factory, thread_naming_pattern, thread_pool, time_service, time_service_interval, timer, timer_handle_non_blocking_tasks, use_direct_memory, use_vthreads, view, who_has_cache, who_has_cache_timeout

Fields inherited from class Protocol

after_creation_hook, down_prot, ergonomics, id, local_addr, log, policies, preview_warning, stack, stats, up_prot

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	TP()

Method Summary

Modifier and Type	Method	Description
boolean	addPhysicalAddressToCache(Address logical_addr, PhysicalAddress physical_addr)
protected boolean	addPhysicalAddressToCache(Address logical_addr, PhysicalAddress physical_addr, boolean overwrite)
void	clearLogicalAddressCache()	Clears the cache.
void	destroy()	This method is called on a JChannel.close().
void	disableDiagnostics()
void	doSend(ByteBuffer buf, Address dest)
Object	down(Event evt)	An event is to be sent down the stack.
Object	down(Message msg)	A message needs to be sent to a single member or all members
<T extends TP> T	enableDiagnostics()
void	evictLogicalAddressCache()
void	evictLogicalAddressCache(boolean force)
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
protected void	fetchPhysicalAddrs(List<Address> missing)
protected Responses	fetchResponsesFromDiscoveryProtocol(List<Address> missing)
protected Collection<PhysicalAddress>	getAllPhysicalAddressesFromCache()
protected abstract PhysicalAddress	getPhysicalAddress()	Returns the physical address of this transport
PhysicalAddress	getPhysicalAddressFromCache(Address logical_addr)
protected void	handleConnect()
protected void	handleDisconnect()
protected void	handleMessageBatch(DataInput in, boolean multicast)
Map<String,String>	handleProbe(String... keys)	Handles a probe.
protected void	handleSingleMessage(DataInput in, boolean multicast)
void	init()	Called after a protocol has been created and before the protocol is started.
void	passBatchUp(MessageBatch batch, boolean perform_cluster_name_matching, boolean discard_own_mcast)
void	passMessageUp(Message msg, boolean perform_cluster_name_matching, boolean multicast, boolean discard_own_mcast)
protected void	processBatch(MessageBatch batch, boolean oob)
void	receive(Address sender, byte[] data, int offset, int length)	Subclasses must call this method when a unicast or multicast message has been received
void	receive(Address sender, DataInput in, int ignoredLength)
protected void	registerLocalAddress(Address addr)	Associates the address with the physical address fetched from the cache
protected void	removeLogicalAddressFromCache(Address logical_addr)
protected void	sendTo(Address dest, ByteBuffer buf)
protected void	sendToAll(ByteBuffer buf)	Fetches the physical addrs for all mbrs and sends the msg to each physical address.
abstract void	sendUnicast(PhysicalAddress dest, ByteBuffer buf)	Send a unicast message to a member.
protected void	sendUnicasts(List<PhysicalAddress> dests, ByteBuffer data)
<T extends Protocol> T	setAddress(Address addr)
protected void	setInAllThreadFactories(String cluster_name, Address local_address, String pattern)
protected void	setSourceAddress(Message msg)	If the sender is null, set our own address.
protected void	setThreadNames()
void	start()	Creates the unicast and multicast sockets and starts the unicast and multicast receiver threads
protected void	startDiagnostics()
void	stop()	Called on a JChannel.disconnect(); stops work (e.g.
protected void	stopDiagnostics()
String[]	supportedKeys()	Returns a list of supported keys
abstract boolean	supportsMulticasting()	Whether hardware multicasting is supported
String	toString()
boolean	unicastDestMismatch(Address dest)
protected void	unsetThreadNames()
protected boolean	versionMatch(short version, Address sender)

Methods inherited from class TPConfig

bundler, clearDifferentClusterCache, clearDifferentVersionCache, clusterName, createBundler, createDiagnosticsHandler, defaultHeaders, dumpTimerTasks, enableAllStats, enableBlockingTimerTasks, enableStats, getAsyncExecutor, getBindAddr, getBindAddress, getBindPort, getBundler, getBundlerClass, getBundlerType, getClusterName, getClusterNameAscii, getDiagnosticsHandler, getDifferentClusterMessages, getDifferentVersionMessages, getExternalAddr, getExternalPort, getLocalPhysicalAddress, getLocalTransport, getLogDiscardMessages, getLogDiscardMessagesVersion, getLogicalAddrCacheExpiration, getLogicalAddrCacheMaxSize, getLogicalAddrCacheReaperInterval, getLogicalAddressCache, getMessageProcessingPolicy, getMessageStats, getNumberOfThreadDumps, getNumMcastMsgsReceived, getNumMcastMsgsSent, getNumRejectedMsgs, getNumThreads, getNumTimerTasks, getNumUcastMsgsReceived, getNumUcastMsgsSent, getPortRange, getReceiveInterfaces, getRTT, getSocketFactory, getSuppressTimeDifferentClusterWarnings, getSuppressTimeDifferentVersionWarnings, getThreadFactory, getThreadNamingPattern, getThreadPool, getTimer, getTimerClass, getTimerThreads, getTimeService, getTimeServiceInterval, getWhoHasCacheTimeout, isLogicalAddressCacheReaperRunning, isReceiveOnAllInterfaces, isTrace, isTrace, localPhysicalAddress, logDiscardMsgs, logDiscardMsgs, logDiscardMsgsVersion, logDiscardMsgsVersion, loggerType, msgProcessingPolicy, msgProcessingPolicy, printLogicalAddressCache, printWhoHasCache, receiveOnAllInterfaces, receiveOnAllInterfaces, registerProbeHandler, removeCancelledTimerTasks, resetStats, sameCluster, setAsyncExecutor, setBindAddr, setBindAddress, setBindPort, setBindToAllInterfaces, setBundler, setBundlerType, setDiagnosticsHandler, setExternalAddr, setExternalPort, setLevel, setLocalTransport, setLocalTransport, setLogDiscardMessages, setLogDiscardMessagesVersion, setLogicalAddrCacheExpiration, setLogicalAddrCacheMaxSize, setLogicalAddrCacheReaperInterval, setMessageProcessingPolicy, setMessageStats, setPortRange, setSocketFactory, setSuppressTimeDifferentClusterWarnings, setSuppressTimeDifferentVersionWarnings, setThreadFactory, setThreadPool, setTimer, setTimeService, setTimeServiceInterval, setWhoHasCacheTimeout, threadNamingPattern, timestamp, unregisterProbeHandler, unregisterProbeHandler, useDirectMemory, useDirectMemory, useVirtualThreads, useVirtualThreads, view

Methods inherited from class Protocol

accept, addPolicy, addr, addr, afterCreationHook, down, getAddress, getComponents, getDownProtocol, getDownServices, getId, getIdsAbove, getLevel, getLog, getName, getPolicies, getProtocolStack, getTransport, getUpProtocol, getUpServices, getValue, isErgonomics, level, parse, policies, previewWarning, previewWarning, providedDownServices, providedUpServices, removePolicy, requiredDownServices, requiredUpServices, resetStatistics, setDownProtocol, setErgonomics, setId, setPolicies, setProtocolStack, setUpProtocol, setValue, statsEnabled, up, up, up

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Field Details

LIST

public static final byte LIST

See Also:

• Constant Field Values

MULTICAST

public static final byte MULTICAST

See Also:

• Constant Field Values

MSG_OVERHEAD

public static final int MSG_OVERHEAD

See Also:

• Constant Field Values

MIN_WAIT_BETWEEN_DISCOVERIES

protected static final long MIN_WAIT_BETWEEN_DISCOVERIES

Constructor Details

TP

protected TP()

Method Details

supportsMulticasting

public abstract boolean supportsMulticasting()

Whether hardware multicasting is supported

getPhysicalAddress

protected abstract PhysicalAddress getPhysicalAddress()

Returns the physical address of this transport

sendUnicast

public abstract void sendUnicast(PhysicalAddress dest,
 ByteBuffer buf)
                          throws Exception

Send a unicast message to a member. Note that the destination address is a *physical*, not a logical address

Parameters:

dest - Must be a non-null physical unicast address (e.g. IpAddress)

buf - The data to be sent. This is not a copy, so don't modify it

Throws:

Exception

toString

public String toString()

Overrides:

toString in class Protocol

setAddress

public <T extends Protocol> T setAddress(Address addr)

Overrides:

setAddress in class Protocol

init

public void init()
          throws Exception

Description copied from class: Protocol

Called after a protocol has been created and before the protocol is started. Attributes are already set. Other protocols are not yet connected and events cannot yet be sent.

Specified by:

init in interface Lifecycle

Overrides:

init in class Protocol

Throws:

Exception - Thrown if protocol cannot be initialized successfully. This will cause the ProtocolStack to fail, so the the channel constructor will throw an exception

start

public void start()
           throws Exception

Creates the unicast and multicast sockets and starts the unicast and multicast receiver threads

Specified by:

start in interface Lifecycle

Overrides:

start in class Protocol

Throws:

Exception - Thrown if protocol cannot be started successfully. This will cause the ProtocolStack to fail, so JChannel.connect(String) will throw an exception

stop

public void stop()

Description copied from class: Protocol

Called on a JChannel.disconnect(); stops work (e.g. by closing multicast socket). Will be called from top to bottom.

Specified by:

stop in interface Lifecycle

Overrides:

stop in class Protocol

destroy

public void destroy()

Description copied from class: Protocol

This method is called on a JChannel.close(). Does some cleanup; after the call, the VM will terminate

Specified by:

destroy in interface Lifecycle

Overrides:

destroy in class Protocol

enableDiagnostics

public <T extends TP> T enableDiagnostics()

disableDiagnostics

public void disableDiagnostics()

handleProbe

public Map<String,String> handleProbe(String... keys)

Description copied from interface: DiagnosticsHandler.ProbeHandler

Handles a probe. For each key that is handled, the key and its result should be in the returned map.

Specified by:

handleProbe in interface DiagnosticsHandler.ProbeHandler

Returns:

A map of sring keys and values. A null return value is permissible.

supportedKeys

public String[] supportedKeys()

Description copied from interface: DiagnosticsHandler.ProbeHandler

Returns a list of supported keys

Specified by:

supportedKeys in interface DiagnosticsHandler.ProbeHandler

down

public Object down(Event evt)

Description copied from class: Protocol

An event is to be sent down the stack. A protocol may want to examine its type and perform some action on it, depending on the event's type. If the event is a message MSG, then the protocol may need to add a header to it (or do nothing at all) before sending it down the stack using down_prot.down().

Overrides:

down in class Protocol

down

public Object down(Message msg)

A message needs to be sent to a single member or all members

Overrides:

down in class Protocol

passMessageUp

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

passBatchUp

public void passBatchUp(MessageBatch batch,
 boolean perform_cluster_name_matching,
 boolean discard_own_mcast)

receive

public void receive(Address sender,
 byte[] data,
 int offset,
 int length)

Subclasses must call this method when a unicast or multicast message has been received

receive

public void receive(Address sender,
 DataInput in,
 int ignoredLength)
             throws Exception

Throws:

Exception

doSend

public void doSend(ByteBuffer buf,
 Address dest)
            throws Exception

Throws:

Exception

unicastDestMismatch

public boolean unicastDestMismatch(Address dest)

addPhysicalAddressToCache

public boolean addPhysicalAddressToCache(Address logical_addr,
 PhysicalAddress physical_addr)

getPhysicalAddressFromCache

public PhysicalAddress getPhysicalAddressFromCache(Address logical_addr)

handleMessageBatch

protected void handleMessageBatch(DataInput in,
 boolean multicast)

handleSingleMessage

protected void handleSingleMessage(DataInput in,
 boolean multicast)

processBatch

protected void processBatch(MessageBatch batch,
 boolean oob)

versionMatch

protected boolean versionMatch(short version,
 Address sender)

sendTo

protected void sendTo(Address dest,
 ByteBuffer buf)
               throws Exception

Throws:

Exception

sendToAll

protected void sendToAll(ByteBuffer buf)
                  throws Exception

Fetches the physical addrs for all mbrs and sends the msg to each physical address. Asks discovery for missing members' physical addresses if needed

Throws:

Exception

sendUnicasts

protected void sendUnicasts(List<PhysicalAddress> dests,
 ByteBuffer data)
                     throws Exception

Throws:

Exception

fetchPhysicalAddrs

protected void fetchPhysicalAddrs(List<Address> missing)

fetchResponsesFromDiscoveryProtocol

protected Responses fetchResponsesFromDiscoveryProtocol(List<Address> missing)

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 send a message on behalf of someone else, e.g. in case of retransmission, when the original sender has crashed.

handleConnect

protected void handleConnect()
                      throws Exception

Throws:

Exception

handleDisconnect

protected void handleDisconnect()

evictLogicalAddressCache

public void evictLogicalAddressCache()

evictLogicalAddressCache

public void evictLogicalAddressCache(boolean force)

registerLocalAddress

protected void registerLocalAddress(Address addr)

Associates the address with the physical address fetched from the cache

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

startDiagnostics

protected void startDiagnostics()
                         throws Exception

Throws:

Exception

stopDiagnostics

protected void stopDiagnostics()

setThreadNames

protected void setThreadNames()

unsetThreadNames

protected void unsetThreadNames()

setInAllThreadFactories

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

addPhysicalAddressToCache

protected boolean addPhysicalAddressToCache(Address logical_addr,
 PhysicalAddress physical_addr,
 boolean overwrite)

getAllPhysicalAddressesFromCache

protected Collection<PhysicalAddress> getAllPhysicalAddressesFromCache()

removeLogicalAddressFromCache

protected void removeLogicalAddressFromCache(Address logical_addr)

clearLogicalAddressCache

public void clearLogicalAddressCache()

Clears the cache. Do not use, this is only for unit testing !