org.mpisws.p2p.transport.rendezvous
Class RendezvousTransportLayerImpl<Identifier,HighIdentifier extends RendezvousContact>

java.lang.Object
  org.mpisws.p2p.transport.rendezvous.RendezvousTransportLayerImpl<Identifier,HighIdentifier>

Type Parameters:

Identifier -

All Implemented Interfaces:

PilotManager<HighIdentifier>, RendezvousTransportLayer<HighIdentifier>, TransportLayer<Identifier,java.nio.ByteBuffer>, TransportLayerCallback<Identifier,java.nio.ByteBuffer>, Destructable

public class RendezvousTransportLayerImpl<Identifier,HighIdentifier extends RendezvousContact>
extends java.lang.Object
implements TransportLayer<Identifier,java.nio.ByteBuffer>, TransportLayerCallback<Identifier,java.nio.ByteBuffer>, PilotManager<HighIdentifier>, RendezvousTransportLayer<HighIdentifier>

The trick here is that this layer is at some level, say InetSocketAddress, but must pass around very High-Level Identifiers, such as a NodeHandle for the rendezvous strategy to do its job, but maybe this can just be the RendezvousContact, and it can be casted. protocol: byte CONNECTOR_SOCKET HighIdentifier target = serializer.deserialize(sib); HighIdentifier opener = serializer.deserialize(sib); int uid = sib.readInt(); byte ACCEPTOR_SOCKET HighIdentifier target = serializer.deserialize(sib); HighIdentifier opener = serializer.deserialize(sib); int uid = sib.readInt(); UDP: We send these messages over TCP/routing unless we believe the Firewall has a temporarilly open port to us (see the following) The firewall will open the port for a few seconds after a message is sent. However it uses an ephemeral port. So, when we receive a message from a new port, we don't know who it's coming from yet, and if it is a ping and we send back a pong, the higher layer will try to send it to the outer-most addr, but this won't work if the node is NATted and didn't configure forwarding. Ex: Nancy is NATted and her external address:port is x:0 because there is no forwarding. Here's what this layer sees x:5000 -> Alice : UDP // we don't know who x:5000 is, and we don't know what kind of message it is Alice -> x:0 : Pong // we need to translate x:0 into x:5000 To accomplish this we'll use a series of Hash tables, and also tag the incoming addr/port on the options of incoming UDP packets. When Liveness or Identity makes an immediate response to UDP packets we will get our tag back, but when normal traffic is sent we need to remember the mapping. Note that there may be several NATted nodes behind the same firewall who all advertise the same bogus port of x:0, so we can't use the external port as a key in the table. We can assume that every sendMessage either has our tag, or a highIdentifier in the tag (set by lower identity), or both If there is both then we'll make a binding of the tag, and the highIdentifier to the port If there is only a tag, then we'll send it to the tag If there is only a highIdentifier, then we may be forced to TCP/route Also, we need to remember how recently we got a UDP from the node, and if it's been too long, assume that the hole has closed, and we need to shift back to TCP (much slower)

Author:

Jeff Hoye

Field Summary

static byte	ACCEPTOR_SOCKET
protected TransportLayerCallback<Identifier,java.nio.ByteBuffer>	callback
static byte	CONNECTION_RESPONSE_FAILURE
static byte	CONNECTION_RESPONSE_SUCCESS
static byte	CONNECTOR_SOCKET
protected ContactDirectStrategy<HighIdentifier>	contactDirectStrategy
protected EphemeralDB<Identifier,HighIdentifier>	ephemeralDB
protected ErrorHandler<Identifier>	errorHandler
static java.lang.String	FROM_OVERLAY The message came from the overlay, rather than a lower layer
protected HighIdentifier	localNodeHandle
protected Logger	logger
static long	NO_TAG
static byte	NORMAL_SOCKET
static java.lang.String	OPTION_USE_PILOT Value should be a HighIdentifier
static byte	PILOT_PING
static byte[]	PILOT_PING_BYTES
static int	PILOT_PING_PERIOD
static byte	PILOT_PONG
static byte[]	PILOT_PONG_BYTES
static byte	PILOT_REQUEST
static byte	PILOT_SOCKET
static byte[]	PILOT_SOCKET_BYTES
protected PilotFinder<HighIdentifier>	pilotFinder
protected RandomSource	random
java.lang.String	RENDEZVOUS_CONTACT_STRING options.get(RENDEZVOUS_CONTACT_STRING) returns a RendezvousContact
protected RendezvousGenerationStrategy<HighIdentifier>	rendezvousGenerator
protected RendezvousStrategy<HighIdentifier>	rendezvousStrategy
protected ResponseStrategy<Identifier>	responseStrategy
protected SelectorManager	selectorManager
protected ContactDeserializer<Identifier,HighIdentifier>	serializer
static java.lang.String	TAG_KEY
protected TimeSource	time
protected TransportLayer<Identifier,java.nio.ByteBuffer>	tl

Fields inherited from interface org.mpisws.p2p.transport.rendezvous.RendezvousTransportLayer

SUCCESS

Constructor Summary

RendezvousTransportLayerImpl(TransportLayer<Identifier,java.nio.ByteBuffer> tl, java.lang.String RENDEZVOUS_CONTACT_STRING, HighIdentifier myRendezvousContact, ContactDeserializer<Identifier,HighIdentifier> deserializer, RendezvousGenerationStrategy<HighIdentifier> rendezvousGenerator, PilotFinder<HighIdentifier> pilotFinder, RendezvousStrategy<HighIdentifier> rendezvousStrategy, ResponseStrategy<Identifier> responseStrategy, ContactDirectStrategy<HighIdentifier> contactDirectStrategy, Environment env)

Method Summary

void	acceptMessages(boolean b) Toggle accepting incoming messages.
void	acceptSockets(boolean b) Toggle accepting new sockets.
void	addIncomingPilotListener(IncomingPilotListener<HighIdentifier> listener)
void	addOutgoingPilotListener(OutgoingPilotListener<HighIdentifier> listener)
void	closePilot(HighIdentifier i) Tells the manager that the pilot to the Identifier is no longer useful
protected void	createForwarder(P2PSocket<Identifier> a, P2PSocket<Identifier> b, HighIdentifier connector, HighIdentifier acceptor, int uid)
void	destroy()
protected HighIdentifier	getHighIdentifier(java.util.Map<java.lang.String,java.lang.Object> options)
Identifier	getLocalIdentifier() The local node.
protected long	getTag(java.util.Map<java.lang.String,java.lang.Object> options)
void	incomingSocket(P2PSocket<Identifier> s) Notification of a new socket.
void	messageReceived(Identifier i, java.nio.ByteBuffer m, java.util.Map<java.lang.String,java.lang.Object> options) Called when a new message is received.
void	messageReceivedFromOverlay(HighIdentifier i, java.nio.ByteBuffer m, java.util.Map<java.lang.String,java.lang.Object> options) Usually called from the higher level app, who probably used routing to get the message here.
protected void	notifyIncomingPilotAdded(HighIdentifier i)
protected void	notifyIncomingPilotRemoved(HighIdentifier i)
protected void	notifyOutgoingPilotAdded(HighIdentifier i)
protected void	notifyOutgoingPilotRemoved(HighIdentifier i)
protected void	openAcceptSocket(HighIdentifier requestor, HighIdentifier middleMan, int uid) We are a firewalled node and got a connect request, now time to respond to it
void	openChannel(HighIdentifier requestor, HighIdentifier middleMan, int uid) Open a socket to the dest, then after writing credentials, call notify the higher layer: incomingSocket()
SocketRequestHandle<HighIdentifier>	openPilot(HighIdentifier i, Continuation<SocketRequestHandle<HighIdentifier>,java.lang.Exception> deliverAckToMe) Only used by NATted node.
SocketRequestHandle<Identifier>	openSocket(Identifier i, SocketCallback<Identifier> deliverSocketToMe, java.util.Map<java.lang.String,java.lang.Object> options) Open a socket to the Identifier
protected boolean	openSocketUsingPilotToMe(HighIdentifier contact, SocketRequestHandle<Identifier> handle, SocketCallback<Identifier> deliverSocketToMe, java.util.Map<java.lang.String,java.lang.Object> options) Return true there was a pilot to me.
protected void	openSocketViaPilot(HighIdentifier dest, HighIdentifier middleMan, SocketRequestHandle<Identifier> handle, SocketCallback<Identifier> deliverSocketToMe, java.util.Map<java.lang.String,java.lang.Object> options)
void	putConnectSocket(HighIdentifier requestor, HighIdentifier target, int uid, P2PSocket<Identifier> socket) This map stores the connect socket until the corresponding accept socket arrives
protected void	putExpectedIncomingSocket(HighIdentifier contact, int uid, SocketCallback<Identifier> deliverSocketToMe, SocketRequestHandle<Identifier> requestHandle)
protected void	readAcceptHeader(P2PSocket<Identifier> acceptorSocket)
protected void	readConnectHeader(P2PSocket<Identifier> socket)
P2PSocket<Identifier>	removeConnectSocket(HighIdentifier requestor, HighIdentifier target, int uid)
protected Tuple<SocketCallback<Identifier>,SocketRequestHandle<Identifier>>	removeExpectedIncomingSocket(HighIdentifier target, int uid)
void	removeIncomingPilotListener(IncomingPilotListener<HighIdentifier> listener)
void	removeOutgoingPilotListener(OutgoingPilotListener<HighIdentifier> listener)
protected void	routeForSocket()
MessageRequestHandle<Identifier,java.nio.ByteBuffer>	sendMessage(Identifier i, java.nio.ByteBuffer m, MessageCallback<Identifier,java.nio.ByteBuffer> deliverAckToMe, java.util.Map<java.lang.String,java.lang.Object> options) What to do if firewalled? ConnectRequest UDP only? For now always use UDP_AND_TCP
void	setCallback(TransportLayerCallback<Identifier,java.nio.ByteBuffer> callback) Set the callback for incoming sockets/messages
void	setErrorHandler(ErrorHandler<Identifier> handler) To be notified of problems not related to an outgoing messaage/socket.
java.lang.String	toString()

Methods inherited from class java.lang.Object

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

Field Detail

NORMAL_SOCKET

public static final byte NORMAL_SOCKET

See Also:

Constant Field Values

CONNECTOR_SOCKET

public static final byte CONNECTOR_SOCKET

See Also:

Constant Field Values

ACCEPTOR_SOCKET

public static final byte ACCEPTOR_SOCKET

See Also:

Constant Field Values

PILOT_SOCKET

public static final byte PILOT_SOCKET

See Also:

Constant Field Values

CONNECTION_RESPONSE_FAILURE

public static final byte CONNECTION_RESPONSE_FAILURE

See Also:

Constant Field Values

CONNECTION_RESPONSE_SUCCESS

public static final byte CONNECTION_RESPONSE_SUCCESS

See Also:

Constant Field Values

NO_TAG

public static final long NO_TAG

See Also:

Constant Field Values

TAG_KEY

public static final java.lang.String TAG_KEY

See Also:

Constant Field Values

FROM_OVERLAY

public static final java.lang.String FROM_OVERLAY

The message came from the overlay, rather than a lower layer

See Also:

Constant Field Values

OPTION_USE_PILOT

public static final java.lang.String OPTION_USE_PILOT

Value should be a HighIdentifier

See Also:

Constant Field Values

RENDEZVOUS_CONTACT_STRING

public java.lang.String RENDEZVOUS_CONTACT_STRING

options.get(RENDEZVOUS_CONTACT_STRING) returns a RendezvousContact

tl

protected TransportLayer<Identifier,java.nio.ByteBuffer> tl

callback

protected TransportLayerCallback<Identifier,java.nio.ByteBuffer> callback

rendezvousGenerator

protected RendezvousGenerationStrategy<HighIdentifier extends RendezvousContact> rendezvousGenerator

pilotFinder

protected PilotFinder<HighIdentifier extends RendezvousContact> pilotFinder

rendezvousStrategy

protected RendezvousStrategy<HighIdentifier extends RendezvousContact> rendezvousStrategy

responseStrategy

protected ResponseStrategy<Identifier> responseStrategy

localNodeHandle

protected HighIdentifier extends RendezvousContact localNodeHandle

logger

protected Logger logger

serializer

protected ContactDeserializer<Identifier,HighIdentifier extends RendezvousContact> serializer

selectorManager

protected SelectorManager selectorManager

random

protected RandomSource random

time

protected TimeSource time

ephemeralDB

protected EphemeralDB<Identifier,HighIdentifier extends RendezvousContact> ephemeralDB

contactDirectStrategy

protected ContactDirectStrategy<HighIdentifier extends RendezvousContact> contactDirectStrategy

errorHandler

protected ErrorHandler<Identifier> errorHandler

PILOT_PING

public static final byte PILOT_PING

See Also:

Constant Field Values

PILOT_PONG

public static final byte PILOT_PONG

See Also:

Constant Field Values

PILOT_REQUEST

public static final byte PILOT_REQUEST

See Also:

Constant Field Values

PILOT_PING_BYTES

public static final byte[] PILOT_PING_BYTES

PILOT_PONG_BYTES

public static final byte[] PILOT_PONG_BYTES

PILOT_SOCKET_BYTES

public static final byte[] PILOT_SOCKET_BYTES

PILOT_PING_PERIOD

public static final int PILOT_PING_PERIOD

See Also:

Constant Field Values

Constructor Detail

RendezvousTransportLayerImpl

public RendezvousTransportLayerImpl(TransportLayer<Identifier,java.nio.ByteBuffer> tl,
                                    java.lang.String RENDEZVOUS_CONTACT_STRING,
                                    HighIdentifier myRendezvousContact,
                                    ContactDeserializer<Identifier,HighIdentifier> deserializer,
                                    RendezvousGenerationStrategy<HighIdentifier> rendezvousGenerator,
                                    PilotFinder<HighIdentifier> pilotFinder,
                                    RendezvousStrategy<HighIdentifier> rendezvousStrategy,
                                    ResponseStrategy<Identifier> responseStrategy,
                                    ContactDirectStrategy<HighIdentifier> contactDirectStrategy,
                                    Environment env)

Method Detail

openSocket

public SocketRequestHandle<Identifier> openSocket(Identifier i,
                                                  SocketCallback<Identifier> deliverSocketToMe,
                                                  java.util.Map<java.lang.String,java.lang.Object> options)

Description copied from interface: TransportLayer

Open a socket to the Identifier

Specified by:

openSocket in interface TransportLayer<Identifier,java.nio.ByteBuffer>

Parameters:

i - who to open the socket to

deliverSocketToMe - the callback when the socket is opened

options - options on how to open the socket (don't source route, encrypt etc) (may not be respected if layer cannot provide service)

Returns:

an object to cancel opening the socket if it takes to long, or is no longer relevant

openSocketUsingPilotToMe

protected boolean openSocketUsingPilotToMe(HighIdentifier contact,
                                           SocketRequestHandle<Identifier> handle,
                                           SocketCallback<Identifier> deliverSocketToMe,
                                           java.util.Map<java.lang.String,java.lang.Object> options)

Return true there was a pilot to me.

Parameters:

contact -

handle -

deliverSocketToMe -

Returns:

openSocketViaPilot

protected void openSocketViaPilot(HighIdentifier dest,
                                  HighIdentifier middleMan,
                                  SocketRequestHandle<Identifier> handle,
                                  SocketCallback<Identifier> deliverSocketToMe,
                                  java.util.Map<java.lang.String,java.lang.Object> options)

routeForSocket

protected void routeForSocket()

incomingSocket

public void incomingSocket(P2PSocket<Identifier> s)
                    throws java.io.IOException

Description copied from interface: TransportLayerCallback

Notification of a new socket.

Specified by:

incomingSocket in interface TransportLayerCallback<Identifier,java.nio.ByteBuffer>

Parameters:

s - the incoming socket

Throws:

java.io.IOException

readConnectHeader

protected void readConnectHeader(P2PSocket<Identifier> socket)
                          throws java.io.IOException

Throws:

java.io.IOException

readAcceptHeader

protected void readAcceptHeader(P2PSocket<Identifier> acceptorSocket)
                         throws java.io.IOException

Throws:

java.io.IOException

putExpectedIncomingSocket

protected void putExpectedIncomingSocket(HighIdentifier contact,
                                         int uid,
                                         SocketCallback<Identifier> deliverSocketToMe,
                                         SocketRequestHandle<Identifier> requestHandle)

removeExpectedIncomingSocket

protected Tuple<SocketCallback<Identifier>,SocketRequestHandle<Identifier>> removeExpectedIncomingSocket(HighIdentifier target,
                                                                                                         int uid)

createForwarder

protected void createForwarder(P2PSocket<Identifier> a,
                               P2PSocket<Identifier> b,
                               HighIdentifier connector,
                               HighIdentifier acceptor,
                               int uid)

putConnectSocket

public void putConnectSocket(HighIdentifier requestor,
                             HighIdentifier target,
                             int uid,
                             P2PSocket<Identifier> socket)

This map stores the connect socket until the corresponding accept socket arrives

Parameters:

socket -

requestor -

target -

uid -

removeConnectSocket

public P2PSocket<Identifier> removeConnectSocket(HighIdentifier requestor,
                                                 HighIdentifier target,
                                                 int uid)

openChannel

public void openChannel(HighIdentifier requestor,
                        HighIdentifier middleMan,
                        int uid)

Description copied from interface: RendezvousTransportLayer

Open a socket to the dest, then after writing credentials, call notify the higher layer: incomingSocket()

Specified by:

openChannel in interface RendezvousTransportLayer<HighIdentifier extends RendezvousContact>

openAcceptSocket

protected void openAcceptSocket(HighIdentifier requestor,
                                HighIdentifier middleMan,
                                int uid)

We are a firewalled node and got a connect request, now time to respond to it

Parameters:

requestor -

i -

sib -

messageReceivedFromOverlay

public void messageReceivedFromOverlay(HighIdentifier i,
                                       java.nio.ByteBuffer m,
                                       java.util.Map<java.lang.String,java.lang.Object> options)
                                throws java.io.IOException

Usually called from the higher level app, who probably used routing to get the message here.

Specified by:

messageReceivedFromOverlay in interface RendezvousTransportLayer<HighIdentifier extends RendezvousContact>

Parameters:

i -

m -

options -

Throws:

java.io.IOException

messageReceived

public void messageReceived(Identifier i,
                            java.nio.ByteBuffer m,
                            java.util.Map<java.lang.String,java.lang.Object> options)
                     throws java.io.IOException

Description copied from interface: TransportLayerCallback

Called when a new message is received.

Specified by:

messageReceived in interface TransportLayerCallback<Identifier,java.nio.ByteBuffer>

Parameters:

i - The node it is coming from

m - the message

options - describe how the message arrived (udp/tcp, encrypted etc)

Throws:

java.io.IOException - if there is a problem decoding the message

getHighIdentifier

protected HighIdentifier getHighIdentifier(java.util.Map<java.lang.String,java.lang.Object> options)

getTag

protected long getTag(java.util.Map<java.lang.String,java.lang.Object> options)

sendMessage

public MessageRequestHandle<Identifier,java.nio.ByteBuffer> sendMessage(Identifier i,
                                                                        java.nio.ByteBuffer m,
                                                                        MessageCallback<Identifier,java.nio.ByteBuffer> deliverAckToMe,
                                                                        java.util.Map<java.lang.String,java.lang.Object> options)

What to do if firewalled? ConnectRequest UDP only? For now always use UDP_AND_TCP

Specified by:

sendMessage in interface TransportLayer<Identifier,java.nio.ByteBuffer>

Parameters:

i - the destination

m - the message

deliverAckToMe - layer dependent notification when the message is sent (can indicate placed on the wire, point-to-point acknowledgment, or end-to-end acknowledgement)

options - delivery options (don't source route, encrypt etc) (may not be respected if layer cannot provide service)

Returns:

ability to cancel the message if no longer relevant

toString

public java.lang.String toString()

Overrides:

toString in class java.lang.Object

acceptMessages

public void acceptMessages(boolean b)

Description copied from interface: TransportLayer

Toggle accepting incoming messages. Useful in flow control if overwhelmed by incoming sockets. Default: true

Specified by:

acceptMessages in interface TransportLayer<Identifier,java.nio.ByteBuffer>

acceptSockets

public void acceptSockets(boolean b)

Description copied from interface: TransportLayer

Toggle accepting new sockets. Useful in flow control if overwhelmed by incoming sockets. Default: true

Specified by:

acceptSockets in interface TransportLayer<Identifier,java.nio.ByteBuffer>

getLocalIdentifier

public Identifier getLocalIdentifier()

Description copied from interface: TransportLayer

The local node.

Specified by:

getLocalIdentifier in interface TransportLayer<Identifier,java.nio.ByteBuffer>

Returns:

The local node.

setCallback

public void setCallback(TransportLayerCallback<Identifier,java.nio.ByteBuffer> callback)

Description copied from interface: TransportLayer

Set the callback for incoming sockets/messages

Specified by:

setCallback in interface TransportLayer<Identifier,java.nio.ByteBuffer>

Parameters:

callback - the callback for incoming sockets/messages

setErrorHandler

public void setErrorHandler(ErrorHandler<Identifier> handler)

Description copied from interface: TransportLayer

To be notified of problems not related to an outgoing messaage/socket. Or to be notified if a callback isn't provided.

Specified by:

setErrorHandler in interface TransportLayer<Identifier,java.nio.ByteBuffer>

Parameters:

handler - to be notified of problems not related to a specific messaage/socket.

destroy

public void destroy()

Specified by:

destroy in interface Destructable

notifyOutgoingPilotAdded

protected void notifyOutgoingPilotAdded(HighIdentifier i)

notifyOutgoingPilotRemoved

protected void notifyOutgoingPilotRemoved(HighIdentifier i)

addOutgoingPilotListener

public void addOutgoingPilotListener(OutgoingPilotListener<HighIdentifier> listener)

Specified by:

addOutgoingPilotListener in interface PilotManager<HighIdentifier extends RendezvousContact>

removeOutgoingPilotListener

public void removeOutgoingPilotListener(OutgoingPilotListener<HighIdentifier> listener)

Specified by:

removeOutgoingPilotListener in interface PilotManager<HighIdentifier extends RendezvousContact>

openPilot

public SocketRequestHandle<HighIdentifier> openPilot(HighIdentifier i,
                                                     Continuation<SocketRequestHandle<HighIdentifier>,java.lang.Exception> deliverAckToMe)

Only used by NATted node. Opens a pilot socket to a "lifeline" node. These are usually nodes near the local node in the id space.

Specified by:

openPilot in interface PilotManager<HighIdentifier extends RendezvousContact>

deliverAckToMe - optional

Returns:

closePilot

public void closePilot(HighIdentifier i)

Description copied from interface: PilotManager

Tells the manager that the pilot to the Identifier is no longer useful

Specified by:

closePilot in interface PilotManager<HighIdentifier extends RendezvousContact>

notifyIncomingPilotAdded

protected void notifyIncomingPilotAdded(HighIdentifier i)

notifyIncomingPilotRemoved

protected void notifyIncomingPilotRemoved(HighIdentifier i)

addIncomingPilotListener

public void addIncomingPilotListener(IncomingPilotListener<HighIdentifier> listener)

Specified by:

addIncomingPilotListener in interface PilotManager<HighIdentifier extends RendezvousContact>

removeIncomingPilotListener

public void removeIncomingPilotListener(IncomingPilotListener<HighIdentifier> listener)

Specified by:

removeIncomingPilotListener in interface PilotManager<HighIdentifier extends RendezvousContact>