net.vattp.data
Class SendThread

java.lang.Object
  |
  +--java.lang.Thread
        |
        +--net.vattp.data.SendThread

All Implemented Interfaces:

Runnable

class SendThread

extends Thread

Field Summary

private static int	HASH_BLOCK_SIZE
private static int	MAX_AGGRAGATION
private Hashtable	myAddressesTried A Hashtable of the InetAddresses tried which failed due to host related problems.
private int	myAggragateLength
private byte[]	myAggragation
private int	myAggragationCount
private DataPath	myDataPath
private boolean	myIsAggragating
private boolean	myIsCompressingMsgLengths
private boolean	myIsDoingHMAC
private boolean	myIsDoingMac
private boolean	myIsStandardCBC
private NetAddr	myLocalAddr
private byte[]	myMACKey
private int	myMacLen
private Object	myNotifications
private OutputStream	myOutputStream
private SynchQueue	myReader
private String	myRemoteAddr
private byte[]	mySequence
private MessageDigest	mySHA1
private Socket	mySocket
private long	myTotalTime
private MsgTransformer	myTransform
private Vat	myVat
private static int	NOTIFY_EVERY Notify DataPath every NOTIFY_EVERY bytes of send progress

Fields inherited from class java.lang.Thread

MAX_PRIORITY, MIN_PRIORITY, NORM_PRIORITY

Constructor Summary

(package private)	SendThread(Socket socket, DataPath path, SynchQueue reader, Vat vat) Make a new SendThread for an incoming socket.
(package private)	SendThread(String remoteAddr, DataPath path, SynchQueue reader, Vat vat, Hashtable addressesTried) Make a new SendThread for an outgoing connection.

Method Summary

private void	addElement(Object elem) Add an element to the aggrated message being constructed.
private void	callDataPath(DataCommThunk thunk) Call a method in our DataPath
private byte[]	computeMAC(byte[] b, int off, int len, byte[] lenField)
private void	flushElements()
private byte[]	getSequence() Get the current Sequence for authentication
private void	increment(byte[] value)
private byte[]	md5Hash(int pad, byte[] data, MessageDigest md5) Calculate the MD5 hash of some data with a specific padding.
(package private) static int	msgLength(int len, byte[] buf, int off, boolean compressed) Calculate the message length field for a message.
private void	noticeProblem(Exception e) Inform the DataPath about a problem
private static void	replicate(byte[] a, byte v) replicate - Copy a single byte to all elements of a byte array.
void	run() Body of the thread.
private void	sendBytes(byte[] b, int off, int len, byte[] lenField, int rawLength) Perform compression, authentication, mac calculation, and send the message
private void	sendBytesWithLength(byte[] b) Send an aggrating mode message with only a single element in the message.
private void	setSHA1() Get an instance of the SHA1 message digest.
private void	writeAndRecordProgress(byte[] b, int off, int len) Write a message in chunks and inform the datapath of progress in the write.
private static void	xor(byte[] a, byte[] b) xor - Exclusive OR two byte arrays.

Methods inherited from class java.lang.Thread

activeCount, checkAccess, countStackFrames, currentThread, destroy, dumpStack, enumerate, getContextClassLoader, getName, getPriority, getThreadGroup, holdsLock, interrupt, interrupted, isAlive, isDaemon, isInterrupted, join, join, join, resume, setContextClassLoader, setDaemon, setName, setPriority, sleep, sleep, start, stop, stop, suspend, toString, yield

Methods inherited from class java.lang.Object

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

Field Detail

NOTIFY_EVERY

private static final int NOTIFY_EVERY

Notify DataPath every NOTIFY_EVERY bytes of send progress

myOutputStream

private OutputStream myOutputStream

mySocket

private Socket mySocket

myLocalAddr

private NetAddr myLocalAddr

myRemoteAddr

private final String myRemoteAddr

myDataPath

private DataPath myDataPath

myReader

private final SynchQueue myReader

myVat

private final Vat myVat

myAddressesTried

private Hashtable myAddressesTried

A Hashtable of the InetAddresses tried which failed due to host related problems. They are not to be tried again in this attempt to connect to the vat. The key is InetAddress, data is InetAddress.

MAX_AGGRAGATION

private static final int MAX_AGGRAGATION

myIsAggragating

private boolean myIsAggragating

myAggragateLength

private int myAggragateLength

myAggragation

private final byte[] myAggragation

myAggragationCount

private int myAggragationCount

myNotifications

private Object myNotifications

myTransform

private MsgTransformer myTransform

myIsStandardCBC

private boolean myIsStandardCBC

HASH_BLOCK_SIZE

private static final int HASH_BLOCK_SIZE

myIsDoingMac

private boolean myIsDoingMac

myIsDoingHMAC

private boolean myIsDoingHMAC

myMACKey

private byte[] myMACKey

mySHA1

private MessageDigest mySHA1

myMacLen

private int myMacLen

mySequence

private byte[] mySequence

myIsCompressingMsgLengths

private boolean myIsCompressingMsgLengths

myTotalTime

private long myTotalTime

Constructor Detail

SendThread

SendThread(String remoteAddr,
           DataPath path,
           SynchQueue reader,
           Vat vat,
           Hashtable addressesTried)

Make a new SendThread for an outgoing connection.

Parameters:

remoteAddr - is the IP:port address to connect to.

path - is the DataPath object that receives notifications as messages are sent and error occur.

reader - is the SynchQueue object to get messages from.

vat - is the Vat object to synchronize with before calling methods in the connection DataPath object

addressesTried - is a Hashtable of the InetAddresses already tried to locate the vat which failed due to host, rather than port related problems, and are not to be tried again.

SendThread

SendThread(Socket socket,
           DataPath path,
           SynchQueue reader,
           Vat vat)

Make a new SendThread for an incoming socket.

Parameters:

socket - is the Socket for the new incoming connection.

path - is the DataPath object that receives notifications as messages are sent and error occur.

reader - is the SynchQueue object to get messages from.

vat - is the Vat object to synchronize with before calling methods in the connection DataPath object

Method Detail

addElement

private void addElement(Object elem)
                 throws IOException

Add an element to the aggrated message being constructed.

Parameters:

elem - is the element to be added.

IOException

callDataPath

private void callDataPath(DataCommThunk thunk)

Call a method in our DataPath

Parameters:

thunk - a Thunk that will perform the call. The thunk will be called after the Vat lock is obtained.

computeMAC

private byte[] computeMAC(byte[] b,
                          int off,
                          int len,
                          byte[] lenField)

flushElements

private void flushElements()
                    throws IOException

IOException

getSequence

private byte[] getSequence()

Get the current Sequence for authentication

increment

private void increment(byte[] value)

xor

private static void xor(byte[] a,
                        byte[] b)

xor - Exclusive OR two byte arrays.

Parameters:

a - is the input and output array.

b - is the array which is XORed with a.

replicate

private static void replicate(byte[] a,
                              byte v)

replicate - Copy a single byte to all elements of a byte array.

Parameters:

a - is the byte array.

v - is the byte to be replicated.

md5Hash

private byte[] md5Hash(int pad,
                       byte[] data,
                       MessageDigest md5)

Calculate the MD5 hash of some data with a specific padding.

The padding allows different values to be obtained from the same data by varing the padding value. We use it to get the different authentication values from the same Diffie Hellman shared secret.

Parameters:

pad - The int (treated as a byte) to be used to pad the MD5 calculation.

data - The data to be hashed.

md5 - The message digest object to be used.

Returns:

A byte array representing the hash.

msgLength

static int msgLength(int len,
                     byte[] buf,
                     int off,
                     boolean compressed)
              throws IOException

Calculate the message length field for a message.

Parameters:

len - is the length of the message to have it's length encoded.

buf - is the output buffer for the encoded length.

off - is the offset in the buffer for where to place the encoded length.

compressed - is true if compressed lengths are to be used.

Returns:

is the offset of the byte after the end of the length.

IOException

noticeProblem

private void noticeProblem(Exception e)

Inform the DataPath about a problem

run

public void run()

Body of the thread. Responsible for dequeueing messages and writing them. Also for dequeueing parameter change orders (e.g. authentication changes) and executing them in order. Calls DataPath.shutdown() before exiting. Will exit if a null is dequeued.

Specified by:

run in interface Runnable

Overrides:

run in class Thread

See Also:

java.lang.Thread#start(), java.lang.Thread#stop(), java.lang.Thread#Thread(java.lang.ThreadGroup, java.lang.Runnable, java.lang.String), java.lang.Runnable#run()

sendBytes

private void sendBytes(byte[] b,
                       int off,
                       int len,
                       byte[] lenField,
                       int rawLength)
                throws IOException

Perform compression, authentication, mac calculation, and send the message

Parameters:

b - The byte array to send

off - The offset in b to send from

len - The length to send

lenField - The compressed length field for the message

rawLength - The length to use for comm statistics

IOException

sendBytesWithLength

private void sendBytesWithLength(byte[] b)
                          throws IOException

Send an aggrating mode message with only a single element in the message.

Parameters:

b - is the byte array which is the data of the message

IOException

setSHA1

private void setSHA1()

Get an instance of the SHA1 message digest.

writeAndRecordProgress

private void writeAndRecordProgress(byte[] b,
                                    int off,
                                    int len)
                             throws IOException

Write a message in chunks and inform the datapath of progress in the write.

IOException