edu.iu.dsc.tws.comms.dfw.io

Class KeyedReceiver

java.lang.Object

edu.iu.dsc.tws.comms.dfw.io.KeyedReceiver

All Implemented Interfaces:

MessageReceiver

Direct Known Subclasses:

DKGatherBatchFinalReceiver, KGatherBatchPartialReceiver, KGatherStreamingReceiver, KReduceBatchReceiver, KReduceStreamingReceiver

public abstract class KeyedReceiver
extends java.lang.Object
implements MessageReceiver

This class is the abstract class that all keyed receivers extend. This provides all the basic data structures and performs initialisation on those data structures.

Field Summary

Fields

Modifier and Type	Field and Description
protected java.util.Map<java.lang.Integer,java.lang.Boolean>	batchDone Tracks if the partial receiver has completed processing for a given target
protected DataFlowOperation	dataFlowOperation The dataflow operation that is related to the class instance.
protected int	destination The destination identifier that defines the next destination for this receiver
protected int	executor Id of the current executor
protected java.util.Map<java.lang.Integer,java.util.Map<java.lang.Integer,java.lang.Boolean>>	finishedSources Map that keeps track of which sources have sent an finished signal.
protected java.util.Map<java.lang.Integer,java.lang.Boolean>	isEmptySent Tracks if the empty message has been sent for each target
protected boolean	isFinalBatchReceiver Indicates whether the receiver instance is a final batch receiver or not.
protected int	keyLimit The number of keys buffered in this receiver before keys are flushed
protected int	limitPerKey The buffer limit for single key.
protected java.util.Map<java.lang.Integer,java.util.Map<java.lang.Object,java.util.Queue<java.lang.Object>>>	messages Map that keeps all the incoming messages to this receiver.
protected int	representSource
protected boolean	representSourceSet
protected java.util.Map<java.lang.Integer,java.util.Queue<java.lang.Object>>	sendQueue Objects that are inserted into this queue are objects that can be sent out from the receiver the progress method removes items from this queue and sends them.
protected java.util.Set<java.lang.Integer>	thisSources

Constructor Summary

Constructors

Constructor and Description
KeyedReceiver()

Method Summary

Modifier and Type	Method and Description
protected boolean	checkIfEmptyIsSent(int target) checks if the Empty message was sent for this target and sends it if not sent and possible to send
protected boolean	finishProgress(boolean needsFurtherProgress, int target) Performs the final steps of the progress method in the receiver.
void	init(Config cfg, DataFlowOperation op, java.util.Map<java.lang.Integer,java.util.List<java.lang.Integer>> expectedIds) Initialize the message receiver with tasks from which messages are expected For each sub edge in graph, for each path, gives the expected task ids
protected boolean	isAllQueuesEmpty(java.util.Queue<java.lang.Object> targetSendQueue) checks if the queue structures used to send data is empty.
boolean	isComplete() Weather we are complete a set of messages
protected boolean	isSourcesFinished(int target) checks if the sources have finished for a given target.
protected boolean	moveMessagesToSendQueue(int target, java.util.Map<java.lang.Object,java.util.Queue<java.lang.Object>> messagesPerTarget) moves all the buffered messages into the sendQueue for the given target
protected boolean	moveMessageToSendQueue(int target, java.util.Map<java.lang.Object,java.util.Queue<java.lang.Object>> messagesPerTarget, java.lang.Object key) Moves all the buffered messages for the given key into the sendQueue and removes the entry in the messages data structure if all the messages are moved
protected boolean	offerMessage(int target, java.lang.Object object) saves the given message (or messages if the object is a list) into the messages data structure if possible and rejects the message if the whole message cannot be added to the messages data structure.
void	onFinish(int source) Once called this method will update the finishedSources data structure so that the given source is marked as finished for each target that is present.
boolean	onMessage(int src, int path, int target, int flags, java.lang.Object object) The actual message callback
boolean	progress() Default progress method for keyed receivers.
protected boolean	sendToTarget(boolean needsFurtherProgress, boolean sourcesFinished, int target, java.util.Queue<java.lang.Object> targetSendQueue) Called from the progress method to perform the communication calls to send the queued messages

Methods inherited from class java.lang.Object

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

Methods inherited from interface edu.iu.dsc.tws.api.comms.messaging.MessageReceiver

clean, close, onMessage

Field Detail

executor

protected int executor

Id of the current executor

limitPerKey

protected int limitPerKey

The buffer limit for single key. Messages are not buffered for a key after this limit is reached. The limit per key is not strictly enforced and the structure may store several more values past this limit for some edge cases.

keyLimit

protected int keyLimit

The number of keys buffered in this receiver before keys are flushed

dataFlowOperation

protected DataFlowOperation dataFlowOperation

The dataflow operation that is related to the class instance. Ex - Reduce, Gather, etc.

destination

protected int destination

The destination identifier that defines the next destination for this receiver

finishedSources

protected java.util.Map<java.lang.Integer,java.util.Map<java.lang.Integer,java.lang.Boolean>> finishedSources

Map that keeps track of which sources have sent an finished signal. The finish signal may either be a SYNC_EMPTY message or a SYNC_MESSAGE message in the message flags. Structure - <target, <source, true/false>>

messages

protected java.util.Map<java.lang.Integer,java.util.Map<java.lang.Object,java.util.Queue<java.lang.Object>>> messages

Map that keeps all the incoming messages to this receiver. A separate map is kept for each target and witin each target values are mapped according to their key value Structure - <target, <key, List<value>>>

sendQueue

protected java.util.Map<java.lang.Integer,java.util.Queue<java.lang.Object>> sendQueue

Objects that are inserted into this queue are objects that can be sent out from the receiver the progress method removes items from this queue and sends them. The onMessage method inserts objects into this queue

batchDone

protected java.util.Map<java.lang.Integer,java.lang.Boolean> batchDone

Tracks if the partial receiver has completed processing for a given target

isEmptySent

protected java.util.Map<java.lang.Integer,java.lang.Boolean> isEmptySent

Tracks if the empty message has been sent for each target

isFinalBatchReceiver

protected boolean isFinalBatchReceiver

Indicates whether the receiver instance is a final batch receiver or not. The value is set to false by default. If the receiver is a final batch receiver the sendQueue data structure is not used the data is always kept in the messages data structure for efficiency

representSource

protected int representSource

thisSources

protected java.util.Set<java.lang.Integer> thisSources

representSourceSet

protected boolean representSourceSet

Constructor Detail

KeyedReceiver

public KeyedReceiver()

Method Detail

init

public void init(Config cfg,
                 DataFlowOperation op,
                 java.util.Map<java.lang.Integer,java.util.List<java.lang.Integer>> expectedIds)

Description copied from interface: MessageReceiver

Initialize the message receiver with tasks from which messages are expected For each sub edge in graph, for each path, gives the expected task ids

target -> source tasks

Specified by:

init in interface MessageReceiver

expectedIds - expected task ids

onMessage

public boolean onMessage(int src,
                         int path,
                         int target,
                         int flags,
                         java.lang.Object object)

Description copied from interface: MessageReceiver

The actual message callback

Specified by:

onMessage in interface MessageReceiver

Parameters:

src - the source task

path - the path that is taken by the message, that is intermediate targets

target - the target of this receiver

flags - the communication flags

object - the actual message

Returns:

true if the message is accepted

offerMessage

protected boolean offerMessage(int target,
                               java.lang.Object object)

saves the given message (or messages if the object is a list) into the messages data structure if possible and rejects the message if the whole message cannot be added to the messages data structure.

Parameters:

target - target for which the messages are to be added

object - the message/messages to be added

Returns:

true if the message was added or false otherwise

onFinish

public void onFinish(int source)

Once called this method will update the finishedSources data structure so that the given source is marked as finished for each target that is present.

Parameters:

source - the task id of the source that has finished

moveMessagesToSendQueue

protected boolean moveMessagesToSendQueue(int target,
                                          java.util.Map<java.lang.Object,java.util.Queue<java.lang.Object>> messagesPerTarget)

moves all the buffered messages into the sendQueue for the given target

Parameters:

target - target for which the move needs to be done

messagesPerTarget - messages for given target

Returns:

true if the messagesPerTarget is not empty at the end of the moving process or false otherwise

moveMessageToSendQueue

protected boolean moveMessageToSendQueue(int target,
                                         java.util.Map<java.lang.Object,java.util.Queue<java.lang.Object>> messagesPerTarget,
                                         java.lang.Object key)

Moves all the buffered messages for the given key into the sendQueue and removes the entry in the messages data structure if all the messages are moved

Parameters:

target - target for which the move needs to be done

messagesPerTarget - messages for given target

key - the key to be moved

Returns:

true if all the messages for that key are moved successfully

isSourcesFinished

protected boolean isSourcesFinished(int target)

checks if the sources have finished for a given target.

Parameters:

target - the target to be checked

Returns:

true if all the sources for the given target are finished or false otherwise

checkIfEmptyIsSent

protected boolean checkIfEmptyIsSent(int target)

checks if the Empty message was sent for this target and sends it if not sent and possible to send

Parameters:

target - target for which the check is done

Returns:

false if Empty is sent

progress

public boolean progress()

Default progress method for keyed receivers. This method is targeted at partial receivers which typically execute the same logic. For custom progress logic this method needs to be overwritten

Specified by:

progress in interface MessageReceiver

Returns:

true if needs further communicationProgress

isComplete

public boolean isComplete()

Description copied from interface: MessageReceiver

Weather we are complete a set of messages

Specified by:

isComplete in interface MessageReceiver

Returns:

true if a round is complete with syncs

finishProgress

protected boolean finishProgress(boolean needsFurtherProgress,
                                 int target)

Performs the final steps of the progress method in the receiver. If the method of finishing needs to be changed this method needs to be overwritten.

Parameters:

needsFurtherProgress - current state of needsFurtherProgress value

target - the target(which is a source in this instance) from which the messages are sent

Returns:

true if further progress is needed or false otherwise

isAllQueuesEmpty

protected boolean isAllQueuesEmpty(java.util.Queue<java.lang.Object> targetSendQueue)

checks if the queue structures used to send data is empty. If Additional data structures are used this method needs to be overwritten to include them

Parameters:

targetSendQueue - message queue for the current target

Returns:

true if all the related queues and structures are empty

sendToTarget

protected boolean sendToTarget(boolean needsFurtherProgress,
                               boolean sourcesFinished,
                               int target,
                               java.util.Queue<java.lang.Object> targetSendQueue)

Called from the progress method to perform the communication calls to send the queued messages

Parameters:

needsFurtherProgress - current state of needsFurtherProgress value

sourcesFinished - specifies if the sources have completed

target - the target(which is a source in this instance) from which the messages are sent

targetSendQueue - the data structure that contains all the message data

Returns:

true if further progress is needed or false otherwise