org.apache.reef.io.network.group.impl.task

Class OperatorTopologyImpl

java.lang.Object

org.apache.reef.io.network.group.impl.task.OperatorTopologyImpl

All Implemented Interfaces:

OperatorTopology

public class OperatorTopologyImpl
extends Object
implements OperatorTopology

Constructor Summary

Constructors

Constructor and Description
OperatorTopologyImpl(Class<? extends Name<String>> groupName, Class<? extends Name<String>> operName, String selfId, String driverId, Sender sender, int version)

Method Summary

Methods

Modifier and Type	Method and Description
void	handle(GroupCommunicationMessage msg) Handle messages meant for this operator.
void	initialize()
byte[]	recvFromChildren()
<T> T	recvFromChildren(Reduce.ReduceFunction<T> redFunc, Codec<T> dataCodec)
byte[]	recvFromParent(ReefNetworkGroupCommProtos.GroupCommMessage.Type msgType)
void	sendToChildren(byte[] data, ReefNetworkGroupCommProtos.GroupCommMessage.Type msgType)
void	sendToChildren(Map<String,byte[]> dataMap, ReefNetworkGroupCommProtos.GroupCommMessage.Type msgType)
void	sendToParent(byte[] data, ReefNetworkGroupCommProtos.GroupCommMessage.Type msgType)

Methods inherited from class java.lang.Object

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

Constructor Detail

OperatorTopologyImpl

public OperatorTopologyImpl(Class<? extends Name<String>> groupName,
                    Class<? extends Name<String>> operName,
                    String selfId,
                    String driverId,
                    Sender sender,
                    int version)

Method Detail

handle

public void handle(GroupCommunicationMessage msg)

Handle messages meant for this operator. Data msgs are passed on to the DataHandlingStage while Ctrl msgs are queued up for the base topology to update later. Ctrl msgs signalling death of a task are also routed to the effectiveTopology in order to notify a waiting operation. During initialization when effective topology is not yet set-up, these *Dead msgs are queued in deletionDeltas for the small time window when these arrive after baseTopology has received TopologySetup but not yet created the effectiveTopology. Most times the msgs in the deletionDeltas will be discarded as stale msgs

No synchronization is needed while handling *Dead messages. There 2 states: UpdatingTopo and NotUpdatingTopo If UpdatingTopo, deltas.put still takes care of adding this msg to effTop through baseTopo changes. If not, we add to effTopo. So we are good.

However, for data msgs synchronization is needed. Look at doc of DataHandlingStage

Adding to deletionDeltas should be outside effTopo!=null block. There is a rare possibility that during initialization just after baseTopo is created(so deltas will be ignored) and just before effTopo is created(so effTopo will be null) where we can miss a deletion msg if not added to deletionDelta because this method is synchronized

Specified by:

handle in interface OperatorTopology

initialize

public void initialize()
                throws ParentDeadException

Specified by:

initialize in interface OperatorTopology

Throws:

ParentDeadException

sendToParent

public void sendToParent(byte[] data,
                ReefNetworkGroupCommProtos.GroupCommMessage.Type msgType)
                  throws ParentDeadException

Specified by:

sendToParent in interface OperatorTopology

Throws:

ParentDeadException

sendToChildren

public void sendToChildren(byte[] data,
                  ReefNetworkGroupCommProtos.GroupCommMessage.Type msgType)
                    throws ParentDeadException

Specified by:

sendToChildren in interface OperatorTopology

Throws:

ParentDeadException

sendToChildren

public void sendToChildren(Map<String,byte[]> dataMap,
                  ReefNetworkGroupCommProtos.GroupCommMessage.Type msgType)
                    throws ParentDeadException

Specified by:

sendToChildren in interface OperatorTopology

Throws:

ParentDeadException

recvFromParent

public byte[] recvFromParent(ReefNetworkGroupCommProtos.GroupCommMessage.Type msgType)
                      throws ParentDeadException

Specified by:

recvFromParent in interface OperatorTopology

Throws:

ParentDeadException

recvFromChildren

public <T> T recvFromChildren(Reduce.ReduceFunction<T> redFunc,
                     Codec<T> dataCodec)
                   throws ParentDeadException

Specified by:

recvFromChildren in interface OperatorTopology

Throws:

ParentDeadException

recvFromChildren

public byte[] recvFromChildren()
                        throws ParentDeadException

Specified by:

recvFromChildren in interface OperatorTopology

Throws:

ParentDeadException