edu.umass.cs.nio

Class NIOTransport<NodeIDType>

java.lang.Object

edu.umass.cs.nio.NIOTransport<NodeIDType>

Type Parameters:

NodeIDType - This class is designed for NIO transport. The nice thing about this class is that all steps are truly asynchronous. Even connect calls. There is no blocking on connect, read, or write anywhere and no polling in the code. The code can be used to send a byte stream to a numbered node (with ID->InetSocketAddress mappings specified using the NodeConfig interface) or to directly send messages to an InetSocketAddress. Received data is processed using the DataProcessingWorker interface. Received data can not be associated with a node ID as the received data just looks like a byte stream to the receiving node. The supplied DataProcessingWorker is expected to know what to do with the received byte streams. The high-level organization of this code has a selector thread that waits for connect, accept, read, or write events on socket channels that could result from accepting connections (server side) or from initiating connections upon send requests by application threads. The selector thread also writes to socket channels and reads incoming data from socket channels and passes them off to DataProcessingWorker. The selector thread is the only thread to touch socket channel selection keys, e.g., to change ops or select() on them. Other application threads may invoke send() concurrently to send data. This data is queued in pendingWrites, a synchronized structure between application threads and the selector thread. A pending write is associated with an InetSocketAddress. So, the selector thread can always re-establish a connection to continue writing the byte stream in case the existing connection fails. The selector thread reads from pendingWrites and sets selection ops to wait for write-ready events accordingly. To enable non-blocking connects, the application threads queue connect events in the synchronized structure pendingConnects. The selector thread reads from pendingConnects and sets selection ops to wait for a connect event as needed. A map SockAddrToSockChannel keeps track of the current socket channel being used to send data to a given InetSocketAddress. Note that this mapping can change if connections fail and are re-established. A failed connection can get re-established on demand by an application thread or by the selector thread when it tries to actually write the data to a socket channel and encounters an exception.

All Implemented Interfaces:

HandshakeCallback, java.lang.Runnable

Direct Known Subclasses:

MessageNIOTransport

public class NIOTransport<NodeIDType>
extends java.lang.Object
implements java.lang.Runnable, HandshakeCallback

Author:

V. Arun

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	NIOTransport.AlternatingByteBuffer

Field Summary

Fields

Modifier and Type	Field and Description
protected static SSLDataProcessingWorker.SSL_MODES	DEFAULT_SSL_MODE A flag to easily enable or disable SSL by default.
static int	MAX_PAYLOAD_SIZE Invoked only by the selector thread.
static int	MAX_QUEUED_SENDS Number of sends that can be queued because the connection was established but the remote end crashed before the send was complete.
static int	MIN_INTER_CONNECT_TIME Milliseconds before reconnection attempts.
protected NodeIDType	myID Usually an id that corresponds to a socket address as specified in NodeConfig.
protected NodeConfig<NodeIDType>	nodeConfig
protected DataProcessingWorker	worker
protected static int	WRITE_BUFFER_SIZE

Constructor Summary

Constructors

Constructor and Description
NIOTransport(java.net.InetAddress address, int port, DataProcessingWorker worker, SSLDataProcessingWorker.SSL_MODES sslMode)
NIOTransport(java.net.InetSocketAddress isa, NodeConfig<NodeIDType> nc, DataProcessingWorker worker, boolean start, SSLDataProcessingWorker.SSL_MODES sslMode)
NIOTransport(int port, DataProcessingWorker worker)
NIOTransport(NodeIDType id, NodeConfig<NodeIDType> nc, DataProcessingWorker worker) The constructor to use for ID-based communication.
NIOTransport(NodeIDType id, NodeConfig<NodeIDType> nc, DataProcessingWorker worker, boolean start, SSLDataProcessingWorker.SSL_MODES sslMode)
NIOTransport(NodeIDType id, NodeConfig<NodeIDType> nc, DataProcessingWorker worker, SSLDataProcessingWorker.SSL_MODES sslMode)

Method Summary

Modifier and Type	Method and Description
protected static void	cleanup(java.nio.channels.SelectionKey key)
static boolean	getCompression()
static int	getCompressionThreshold()
protected java.net.InetAddress	getListeningAddress()
protected int	getListeningPort()
java.net.InetSocketAddress	getListeningSocketAddress()
static java.util.logging.Logger	getLogger()
protected java.net.InetAddress	getNodeAddress()
protected int	getNodePort()
protected static int	getPayloadLength(java.nio.ByteBuffer buf)
protected int	getPendingSize()
SSLDataProcessingWorker.SSL_MODES	getSSLMode()
void	handshakeComplete(java.nio.channels.SelectionKey key)
boolean	isDisconnected(NodeIDType node)
protected boolean	isStarted()
boolean	isStopped()
static void	main(java.lang.String[] args)
protected static boolean	outOfRange(int length)
void	run()
int	send(java.net.InetSocketAddress isa, byte[] data)
int	send(java.net.InetSocketAddress isa, byte[] data, int batchSize) For performance testing.
int	send(NodeIDType id, byte[] data) send() methods are called by external application threads.
static void	setCompression(boolean b)
static void	setCompressionThreshold(int t)
void	setMaxQueuedSends(int maxQ)
void	setMinInterConnectTime(int minInterConnectTime)
NIOTransport<NodeIDType>	setName(java.lang.String name)
static void	setUseSenderTask(boolean b)
void	stop() To close NIOTransport instances gracefully.
java.lang.String	toString()

Methods inherited from class java.lang.Object

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

Field Detail

MAX_QUEUED_SENDS

public static final int MAX_QUEUED_SENDS

Number of sends that can be queued because the connection was established but the remote end crashed before the send was complete. Note that this is the number of packets, not bytes. Currently, there is no way to set a byte limit for the queue size.

See Also:

Constant Field Values

MIN_INTER_CONNECT_TIME

public static final int MIN_INTER_CONNECT_TIME

Milliseconds before reconnection attempts.

See Also:

Constant Field Values

WRITE_BUFFER_SIZE

protected static final int WRITE_BUFFER_SIZE

See Also:

Constant Field Values

myID

protected final NodeIDType myID

Usually an id that corresponds to a socket address as specified in NodeConfig. Note that myID can also null be which means wildcard address or it can be a InetSocket address in the case of Local Name Servers

worker

protected final DataProcessingWorker worker

nodeConfig

protected final NodeConfig<NodeIDType> nodeConfig

DEFAULT_SSL_MODE

protected static final SSLDataProcessingWorker.SSL_MODES DEFAULT_SSL_MODE

A flag to easily enable or disable SSL by default. CLEAR is not really a legitimate SSL mode, i.e., it is not supported by SSLDataProcessingWorker, but we have it here for ease of changing the default.

MAX_PAYLOAD_SIZE

public static final int MAX_PAYLOAD_SIZE

Invoked only by the selector thread. read() is easy as it just needs to read whatever is available and send it off to DataProcessingWorker (that has to deal with the complexity of parsing a byte stream).

See Also:

Constant Field Values

Constructor Detail

NIOTransport

public NIOTransport(NodeIDType id,
                    NodeConfig<NodeIDType> nc,
                    DataProcessingWorker worker)
             throws java.io.IOException

The constructor to use for ID-based communication.

Parameters:

id -

nc -

worker -

Throws:

java.io.IOException

NIOTransport

public NIOTransport(NodeIDType id,
                    NodeConfig<NodeIDType> nc,
                    DataProcessingWorker worker,
                    boolean start,
                    SSLDataProcessingWorker.SSL_MODES sslMode)
             throws java.io.IOException

Parameters:

id -

nc -

worker -

start -

sslMode -

Throws:

java.io.IOException

NIOTransport

public NIOTransport(java.net.InetSocketAddress isa,
                    NodeConfig<NodeIDType> nc,
                    DataProcessingWorker worker,
                    boolean start,
                    SSLDataProcessingWorker.SSL_MODES sslMode)
             throws java.io.IOException

Parameters:

isa -

nc -

worker -

start -

sslMode -

Throws:

java.io.IOException

NIOTransport

public NIOTransport(NodeIDType id,
                    NodeConfig<NodeIDType> nc,
                    DataProcessingWorker worker,
                    SSLDataProcessingWorker.SSL_MODES sslMode)
             throws java.io.IOException

Parameters:

id -

nc -

worker -

sslMode -

Throws:

java.io.IOException

NIOTransport

public NIOTransport(int port,
                    DataProcessingWorker worker)
             throws java.io.IOException

Parameters:

port -

worker -

Throws:

java.io.IOException

NIOTransport

public NIOTransport(java.net.InetAddress address,
                    int port,
                    DataProcessingWorker worker,
                    SSLDataProcessingWorker.SSL_MODES sslMode)
             throws java.io.IOException

Parameters:

address -

port -

worker -

sslMode -

Throws:

java.io.IOException

Method Detail

setMaxQueuedSends

public void setMaxQueuedSends(int maxQ)

Parameters:

maxQ - Refer MAX_QUEUED_SENDS.

setMinInterConnectTime

public void setMinInterConnectTime(int minInterConnectTime)

Parameters:

minInterConnectTime - Refer MIN_INTER_CONNECT_TIME.

getLogger

public static final java.util.logging.Logger getLogger()

Returns:

java.util.logging.Logger used by NIOTransport.

setName

public NIOTransport<NodeIDType> setName(java.lang.String name)

Parameters:

name -

Returns:

this

send

public int send(NodeIDType id,
                byte[] data)
         throws java.io.IOException

send() methods are called by external application threads. They may initiate a connection if one is not available. However, the connection is finished in a non-blocking manner by the selector thread. Data to be sent is queued in pendingWrites, which is read later by the selector thread.

Parameters:

id -

data -

Returns:

Number of bytes sent.

Throws:

java.io.IOException

send

public int send(java.net.InetSocketAddress isa,
                byte[] data)
         throws java.io.IOException

Parameters:

isa -

data -

Returns:

Number of bytes sent.

Throws:

java.io.IOException

send

public int send(java.net.InetSocketAddress isa,
                byte[] data,
                int batchSize)
         throws java.io.IOException

For performance testing. Repeats data batchSize number of times in order to simulate the performance of batched sends.

Parameters:

isa -

data -

batchSize -

Returns:

Number of bytes written.

Throws:

java.io.IOException

getPayloadLength

protected static final int getPayloadLength(java.nio.ByteBuffer buf)
                                     throws java.io.IOException

Throws:

java.io.IOException

run

public void run()

Specified by:

run in interface java.lang.Runnable

isStarted

protected boolean isStarted()

stop

public void stop()

To close NIOTransport instances gracefully.

isStopped

public boolean isStopped()

Returns:

Whether stopped.

getNodeAddress

protected java.net.InetAddress getNodeAddress()

getNodePort

protected int getNodePort()

isDisconnected

public boolean isDisconnected(NodeIDType node)

Parameters:

node -

Returns:

Whether node got disconnected.

setCompression

public static final void setCompression(boolean b)

Parameters:

b -

getCompression

public static final boolean getCompression()

Returns:

True if compression enabled.

setCompressionThreshold

public static final void setCompressionThreshold(int t)

Parameters:

t -

getCompressionThreshold

public static final int getCompressionThreshold()

Returns:

Compression threshold

setUseSenderTask

public static final void setUseSenderTask(boolean b)

Parameters:

b -

getSSLMode

public SSLDataProcessingWorker.SSL_MODES getSSLMode()

Returns:

SSL mode used by this NIO transport.

cleanup

protected static final void cleanup(java.nio.channels.SelectionKey key)

getListeningSocketAddress

public java.net.InetSocketAddress getListeningSocketAddress()

Returns:

Actual listening socket address that in general may be different from that corresponding to myID in nodeConfig because of the wildcard fallback in NIOTransport.

getListeningPort

protected int getListeningPort()

getListeningAddress

protected java.net.InetAddress getListeningAddress()

outOfRange

protected static final boolean outOfRange(int length)

toString

public java.lang.String toString()

Overrides:

toString in class java.lang.Object

handshakeComplete

public void handshakeComplete(java.nio.channels.SelectionKey key)

Specified by:

handshakeComplete in interface HandshakeCallback

getPendingSize

protected int getPendingSize()

main

public static void main(java.lang.String[] args)

Parameters:

args -