org.nd4j.parameterserver.distributed.v2.transport.impl

Class AeronUdpTransport

java.lang.Object

org.nd4j.parameterserver.distributed.v2.transport.impl.BaseTransport

org.nd4j.parameterserver.distributed.v2.transport.impl.AeronUdpTransport

All Implemented Interfaces:

AutoCloseable, Transport

public class AeronUdpTransport
extends BaseTransport
implements AutoCloseable

This class is a UDP implementation of Transport interface, based on Aeron

Author:

raver119@gmail.com

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	AeronUdpTransport.RemoteConnection

Nested classes/interfaces inherited from class org.nd4j.parameterserver.distributed.v2.transport.impl.BaseTransport

BaseTransport.HeartbeatThread, BaseTransport.MessageFlow<T>

Field Summary

Fields

Modifier and Type	Field and Description
protected io.aeron.Aeron	aeron
protected ReentrantLock	aeronLock
protected AtomicBoolean	connectedFlag
protected io.aeron.Aeron.Context	context
protected io.aeron.driver.MediaDriver	driver
protected Map<String,MessageCallable>	interceptors
protected int	MESSAGE_THREADS
protected io.aeron.FragmentAssembler	messageHandler
protected BlockingQueue<VoidMessage>	messageQueue
protected ExecutorService	messagesExecutorService
protected io.aeron.Subscription	ownSubscription
protected Map<String,MessageCallable>	precursors
protected BlockingQueue<INDArrayMessage>	propagationQueue
protected Map<String,AeronUdpTransport.RemoteConnection>	remoteConnections
protected int	SENDER_THREADS
protected AtomicBoolean	shutdownFlag
protected int	SUBSCRIPTION_THREADS
protected Thread	subscriptionThread

Fields inherited from class org.nd4j.parameterserver.distributed.v2.transport.impl.BaseTransport

consumers, executorService, handshakeFlag, historyHolder, id, incomingFlow, masterMode, mesh, meshBuildMode, numerOfNodes, outgoingFlow, replies, restartCallback, rootId, splitter, voidConfiguration

Constructor Summary

Constructors

Constructor and Description
AeronUdpTransport(@NonNull String ownIp, int ownPort, @NonNull String rootIp, int rootPort, @NonNull VoidConfiguration configuration)
AeronUdpTransport(@NonNull String rootIp, int rootPort, @NonNull VoidConfiguration configuration) This constructor creates root transport instance
AeronUdpTransport(@NonNull String ownIp, @NonNull String rootIp, @NonNull VoidConfiguration configuration)

Method Summary

Modifier and Type	Method and Description
protected void	addConnection(@NonNull String ipAndPort)
<T extends VoidMessage> void	addInterceptor(@NonNull Class<T> cls, @NonNull MessageCallable<T> callable) This method add interceptor for incoming messages.
<T extends VoidMessage> void	addPrecursor(@NonNull Class<T> cls, @NonNull MessageCallable<T> callable) This method add precursor for incoming messages.
void	close()
protected void	createSubscription()
void	ensureConnection(String id) This method checks connection to the given node ID, and if it's not connected - establishes connection
protected MeshOrganizer	getMesh() This method returns Mesh stored in this Transport instance PLEASE NOTE: This method is suited for tests
String	id() This method returns id of the current transport
boolean	isConnected() This method checks if all connections required for work are established
protected void	jointMessageHandler(org.agrona.DirectBuffer buffer, int offset, int length, io.aeron.logbuffer.Header header) This method converts aeron buffer into VoidMessage and puts into temp queue for further processing
void	launch() This method starts this Transport instance
void	launchAsMaster() This method will start this Transport instance
void	onMeshUpdate(MeshOrganizer mesh) This method will be called if mesh update was received PLEASE NOTE: This method will be called ONLY if new mesh differs from current one
void	onRemap(String id) This method will be called upon remap request
void	processMessage(@NonNull VoidMessage message) This method will be invoked for all incoming messages PLEASE NOTE: this method is mostly suited for tests
protected void	propagateArrayMessage(INDArrayMessage message, PropagationMode mode)
protected void	redirectedPropagateArrayMessage(INDArrayMessage message)
void	sendMessage(@NonNull VoidMessage message, @NonNull String id) This method will send message to the node specified by Id
void	shutdown() This method shuts down this Transport instance
protected void	shutdownSilent()

Methods inherited from class org.nd4j.parameterserver.distributed.v2.transport.impl.BaseTransport

addRequestConsumer, getRandomDownstreamFrom, getRootId, getUpstreamId, incomingPublisher, internalProcessMessage, isIntroduced, isLoopedNode, outgoingConsumer, propagateBroadcastableMessage, propagateMessage, propagateMessageDirect, sendMessageBlocking, sendMessageBlocking, setRestartCallback, totalNumberOfNodes

Methods inherited from class java.lang.Object

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

Field Detail

interceptors

protected Map<String,MessageCallable> interceptors

precursors

protected Map<String,MessageCallable> precursors

remoteConnections

protected Map<String,AeronUdpTransport.RemoteConnection> remoteConnections

SENDER_THREADS

protected final int SENDER_THREADS

See Also:

Constant Field Values

MESSAGE_THREADS

protected final int MESSAGE_THREADS

See Also:

Constant Field Values

SUBSCRIPTION_THREADS

protected final int SUBSCRIPTION_THREADS

See Also:

Constant Field Values

aeron

protected io.aeron.Aeron aeron

context

protected io.aeron.Aeron.Context context

ownSubscription

protected io.aeron.Subscription ownSubscription

messageHandler

protected io.aeron.FragmentAssembler messageHandler

subscriptionThread

protected Thread subscriptionThread

driver

protected io.aeron.driver.MediaDriver driver

messageQueue

protected BlockingQueue<VoidMessage> messageQueue

propagationQueue

protected BlockingQueue<INDArrayMessage> propagationQueue

aeronLock

protected ReentrantLock aeronLock

shutdownFlag

protected final AtomicBoolean shutdownFlag

connectedFlag

protected final AtomicBoolean connectedFlag

messagesExecutorService

protected ExecutorService messagesExecutorService

Constructor Detail

AeronUdpTransport

public AeronUdpTransport(@NonNull
                         @NonNull String ownIp,
                         @NonNull
                         @NonNull String rootIp,
                         @NonNull
                         @NonNull VoidConfiguration configuration)

AeronUdpTransport

public AeronUdpTransport(@NonNull
                         @NonNull String rootIp,
                         int rootPort,
                         @NonNull
                         @NonNull VoidConfiguration configuration)

This constructor creates root transport instance

Parameters:

rootIp -

rootPort -

configuration -

AeronUdpTransport

public AeronUdpTransport(@NonNull
                         @NonNull String ownIp,
                         int ownPort,
                         @NonNull
                         @NonNull String rootIp,
                         int rootPort,
                         @NonNull
                         @NonNull VoidConfiguration configuration)

Method Detail

createSubscription

protected void createSubscription()

propagateArrayMessage

protected void propagateArrayMessage(INDArrayMessage message,
                                     PropagationMode mode)
                              throws IOException

Overrides:

propagateArrayMessage in class BaseTransport

Throws:

IOException

redirectedPropagateArrayMessage

protected void redirectedPropagateArrayMessage(INDArrayMessage message)
                                        throws IOException

Throws:

IOException

jointMessageHandler

protected void jointMessageHandler(org.agrona.DirectBuffer buffer,
                                   int offset,
                                   int length,
                                   io.aeron.logbuffer.Header header)

This method converts aeron buffer into VoidMessage and puts into temp queue for further processing

Parameters:

buffer -

offset -

length -

header -

onRemap

public void onRemap(String id)

Description copied from interface: Transport

This method will be called upon remap request

Specified by:

onRemap in interface Transport

Overrides:

onRemap in class BaseTransport

ensureConnection

public void ensureConnection(String id)

Description copied from interface: Transport

This method checks connection to the given node ID, and if it's not connected - establishes connection

Specified by:

ensureConnection in interface Transport

Overrides:

ensureConnection in class BaseTransport

addConnection

protected void addConnection(@NonNull
                             @NonNull String ipAndPort)

close

public void close()
           throws Exception

Specified by:

close in interface AutoCloseable

Throws:

Exception

launch

public void launch()

Description copied from interface: Transport

This method starts this Transport instance

Specified by:

launch in interface Transport

Overrides:

launch in class BaseTransport

launchAsMaster

public void launchAsMaster()

Description copied from interface: Transport

This method will start this Transport instance

Specified by:

launchAsMaster in interface Transport

Overrides:

launchAsMaster in class BaseTransport

id

public String id()

Description copied from interface: Transport

This method returns id of the current transport

Specified by:

id in interface Transport

Returns:

isConnected

public boolean isConnected()

Description copied from interface: Transport

This method checks if all connections required for work are established

Specified by:

isConnected in interface Transport

Overrides:

isConnected in class BaseTransport

Returns:

true

sendMessage

public void sendMessage(@NonNull
                        @NonNull VoidMessage message,
                        @NonNull
                        @NonNull String id)

Description copied from interface: Transport

This method will send message to the node specified by Id

Specified by:

sendMessage in interface Transport

shutdownSilent

protected void shutdownSilent()

shutdown

public void shutdown()

Description copied from interface: Transport

This method shuts down this Transport instance

Specified by:

shutdown in interface Transport

Overrides:

shutdown in class BaseTransport

onMeshUpdate

public void onMeshUpdate(MeshOrganizer mesh)

Description copied from interface: Transport

This method will be called if mesh update was received PLEASE NOTE: This method will be called ONLY if new mesh differs from current one

Specified by:

onMeshUpdate in interface Transport

Overrides:

onMeshUpdate in class BaseTransport

addInterceptor

public <T extends VoidMessage> void addInterceptor(@NonNull
                                                   @NonNull Class<T> cls,
                                                   @NonNull
                                                   @NonNull MessageCallable<T> callable)

This method add interceptor for incoming messages. If interceptor is defined for given message class - runnable will be executed instead of processMessage()

Parameters:

cls -

callable -

addPrecursor

public <T extends VoidMessage> void addPrecursor(@NonNull
                                                 @NonNull Class<T> cls,
                                                 @NonNull
                                                 @NonNull MessageCallable<T> callable)

This method add precursor for incoming messages. If precursor is defined for given message class - runnable will be executed before processMessage()

Parameters:

cls -

callable -

processMessage

public void processMessage(@NonNull
                           @NonNull VoidMessage message)

Description copied from interface: Transport

This method will be invoked for all incoming messages PLEASE NOTE: this method is mostly suited for tests

Specified by:

processMessage in interface Transport

Overrides:

processMessage in class BaseTransport

getMesh

protected MeshOrganizer getMesh()

This method returns Mesh stored in this Transport instance PLEASE NOTE: This method is suited for tests

Returns: