org.apache.catalina.cluster.tcp
Class FastAsyncSocketSender

java.lang.Object
  org.apache.catalina.cluster.tcp.DataSender
      org.apache.catalina.cluster.tcp.FastAsyncSocketSender

All Implemented Interfaces:

IDataSender

public class FastAsyncSocketSender

extends DataSender

Send cluster messages from a Message queue with only one socket. Ack and keep Alive Handling is supported. Fast Queue can limit queue size and consume all messages at queue at one block.
Limit the queue lock contention under high load!

• With autoConnect=false at ReplicationTransmitter, you can disconnect the sender and all messages are queued. Only use this for small maintaince isuses!
• waitForAck=true, means that receiver ack the transfer
• after one minute idle time, or number of request (100) the connection is reconnected with next request. Change this for production use!
• default ackTimeout is 15 sec: this is very low for big all session replication messages after restart a node
• disable keepAlive: keepAliveTimeout="-1" and keepAliveMaxRequestCount="-1"
• maxQueueLength: Limit the sender queue length (membership goes well, but transfer is failure!!)

FIXME: refactor code duplications with AsyncSocketSender => configurable or extract super class

Since:

5.5.9

Version:

$Id: FastAsyncSocketSender.java 939539 2010-04-30 01:31:33Z kkolinko $

Author:

Peter Rossbach ( idea comes form Rainer Jung)

Field Summary

Fields inherited from class org.apache.catalina.cluster.tcp.DataSender

connectCounter, createTime, dataFailureCounter, dataResendCounter, disconnectCounter, doProcessingStats, doWaitAckStats, keepAliveConnectTime, keepAliveCount, maxProcessingTime, maxWaitAckTime, minProcessingTime, minWaitAckTime, missingAckCounter, nrOfRequests, processingTime, sm, totalBytes, waitAckTime

Constructor Summary

FastAsyncSocketSender(java.lang.String domain, java.net.InetAddress host, int port)
start background thread to push incomming cluster messages to replication node

Method Summary

protected void	checkThread() Start Queue thread as daemon
void	connect() Connect to socket and start background thread to push queued messages
void	disconnect() Disconnect socket ad stop queue thread
java.lang.String	getInfo() Return descriptive information about this implementation and the corresponding version number, in the format <description>/<version>.
long	getInQueueCounter()
int	getMaxQueueLength()
long	getOutQueueCounter()
long	getQueueAddWaitTime()
long	getQueueAddWaitTimeout() get current add wait timeout
long	getQueuedNrOfBytes()
long	getQueueRemoveWaitTime()
long	getQueueRemoveWaitTimeout() get current remove wait timeout
int	getQueueSize()
int	getRecoverCounter() get current push message recover counter
long	getRecoverTimeout() get current push message recover timeout
int	getThreadPriority() Get the current threadPriority
boolean	isQueueCheckLock()
boolean	isQueueDoStats()
boolean	isQueueTimeWait()
void	resetStatistics() Reset sender statistics
void	sendMessage(ClusterData data) Send message to queue for later sending.
void	setMaxQueueLength(int length)
void	setQueueAddWaitTimeout(long timeout) Set add wait timeout (default 10000 msec)
void	setQueueCheckLock(boolean checkLock)
void	setQueueDoStats(boolean doStats)
void	setQueueTimeWait(boolean timeWait)
void	setRecoverCounter(int counter) Set recover couner (default 5 )
void	setRecoverTimeout(long timeout) Set recover timeout (default 5000 msec)
void	setRemoveWaitTimeout(long timeout) set remove wait timeout ( default 30000 msec)
void	setThreadPriority(int threadPriority) change active the queue Thread priority
protected void	stopThread() stop queue worker thread
java.lang.String	toString() Name of this SockerSender

Methods inherited from class org.apache.catalina.cluster.tcp.DataSender

addProcessingStats, addStats, addWaitAckStats, checkKeepAlive, closeSocket, createSocket, getAckTimeout, getAddress, getAvgMessageSize, getAvgProcessingTime, getAvgWaitAckTime, getConnectCounter, getCreateTime, getDataFailureCounter, getDataResendCounter, getDisconnectCounter, getDomain, getKeepAliveConnectTime, getKeepAliveCount, getKeepAliveMaxRequestCount, getKeepAliveTimeout, getMaxProcessingTime, getMaxWaitAckTime, getMinProcessingTime, getMinWaitAckTime, getMissingAckCounter, getNrOfRequests, getPort, getProcessingTime, getRxBufSize, getSenderState, getSocket, getSocketCloseCounter, getSocketOpenCounter, getSocketOpenFailureCounter, getSoLingerTime, getSoTrafficClass, getSuspect, getTimeout, getTotalBytes, getTxBufSize, getWaitAckTime, isConnected, isDoProcessingStats, isDoWaitAckStats, isMessageTransferStarted, isOoBInline, isResend, isSoKeepAlive, isSoLingerOn, isSoReuseAddress, isSuspect, isTcpNoDelay, isWaitForAck, openSocket, pushMessage, setAckTimeout, setAddress, setDomain, setDoProcessingStats, setDoWaitAckStats, setKeepAliveMaxRequestCount, setKeepAliveTimeout, setOoBInline, setPort, setResend, setRxBufSize, setSocket, setSocketConnected, setSoKeepAlive, setSoLingerOn, setSoLingerTime, setSoReuseAddress, setSoTrafficClass, setSuspect, setTcpNoDelay, setTimeout, setTxBufSize, setWaitForAck, waitForAck, writeData

Methods inherited from class java.lang.Object

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

Constructor Detail

FastAsyncSocketSender

public FastAsyncSocketSender(java.lang.String domain,
                             java.net.InetAddress host,
                             int port)

start background thread to push incomming cluster messages to replication node

Parameters:

domain - replication cluster domain (session domain)

host - replication node tcp address

port - replication node tcp port

Method Detail

getInfo

public java.lang.String getInfo()

Return descriptive information about this implementation and the corresponding version number, in the format <description>/<version>.

Overrides:

getInfo in class DataSender

getQueueAddWaitTimeout

public long getQueueAddWaitTimeout()

get current add wait timeout

Returns:

current wait timeout

setQueueAddWaitTimeout

public void setQueueAddWaitTimeout(long timeout)

Set add wait timeout (default 10000 msec)

Parameters:

timeout -

getQueueRemoveWaitTimeout

public long getQueueRemoveWaitTimeout()

get current remove wait timeout

Returns:

The timeout

setRemoveWaitTimeout

public void setRemoveWaitTimeout(long timeout)

set remove wait timeout ( default 30000 msec)

Parameters:

timeout -

isQueueCheckLock

public boolean isQueueCheckLock()

Returns:

Returns the checkLock.

setQueueCheckLock

public void setQueueCheckLock(boolean checkLock)

Parameters:

checkLock - The checkLock to set.

isQueueDoStats

public boolean isQueueDoStats()

Returns:

Returns the doStats.

setQueueDoStats

public void setQueueDoStats(boolean doStats)

Parameters:

doStats - The doStats to set.

isQueueTimeWait

public boolean isQueueTimeWait()

Returns:

Returns the timeWait.

setQueueTimeWait

public void setQueueTimeWait(boolean timeWait)

Parameters:

timeWait - The timeWait to set.

getMaxQueueLength

public int getMaxQueueLength()

Returns:

Returns the inQueueCounter.

setMaxQueueLength

public void setMaxQueueLength(int length)

Parameters:

length - max queue length

getQueueAddWaitTime

public long getQueueAddWaitTime()

Returns:

Returns the add wait times.

getQueueRemoveWaitTime

public long getQueueRemoveWaitTime()

Returns:

Returns the add wait times.

getInQueueCounter

public long getInQueueCounter()

Returns:

Returns the inQueueCounter.

getOutQueueCounter

public long getOutQueueCounter()

Returns:

Returns the outQueueCounter.

getQueueSize

public int getQueueSize()

Returns:

Returns the queueSize.

getRecoverTimeout

public long getRecoverTimeout()

get current push message recover timeout

Returns:

current push message recover timeout

setRecoverTimeout

public void setRecoverTimeout(long timeout)

Set recover timeout (default 5000 msec)

Parameters:

timeout -

getRecoverCounter

public int getRecoverCounter()

get current push message recover counter

Returns:

current push message recover counter

setRecoverCounter

public void setRecoverCounter(int counter)

Set recover couner (default 5 )

Parameters:

counter -

setThreadPriority

public void setThreadPriority(int threadPriority)

change active the queue Thread priority

Parameters:

threadPriority - value must be between MIN and MAX Thread Priority

Throws:

java.lang.IllegalArgumentException

getThreadPriority

public int getThreadPriority()

Get the current threadPriority

Returns:

The thread priority

getQueuedNrOfBytes

public long getQueuedNrOfBytes()

Returns:

Returns the queuedNrOfBytes.

connect

public void connect()
             throws java.io.IOException

Connect to socket and start background thread to push queued messages

Specified by:

connect in interface IDataSender

Overrides:

connect in class DataSender

Throws:

java.io.IOException

See Also:

IDataSender.connect()

disconnect

public void disconnect()

Disconnect socket ad stop queue thread

Specified by:

disconnect in interface IDataSender

Overrides:

disconnect in class DataSender

See Also:

IDataSender.disconnect()

sendMessage

public void sendMessage(ClusterData data)
                 throws java.io.IOException

Send message to queue for later sending.

Specified by:

sendMessage in interface IDataSender

Overrides:

sendMessage in class DataSender

Throws:

java.io.IOException

See Also:

DataSender.pushMessage(ClusterData)

resetStatistics

public void resetStatistics()

Reset sender statistics

Overrides:

resetStatistics in class DataSender

toString

public java.lang.String toString()

Name of this SockerSender

Overrides:

toString in class DataSender

checkThread

protected void checkThread()

Start Queue thread as daemon

stopThread

protected void stopThread()

stop queue worker thread