Giraph源码分析（三）—— 消息通信

Giraph源码分析（三）—— 消息通信

由前文知道每个BSPServiceWorker有一个WorkerServer对象，WorkerServer对象里面又有ServerData对象，作为数据实。ServerData中包含该Worker的partitionStore、edgeStore、incomingMessageStore、currentMessageStore、聚集值等。其中incomingMessageStore对象为MessageStoreByPartition(接口)类型，也就是说消息时按照分区来存储的。MessageStoreByPartition接口的关系图如下：

在SimpleMessageStore抽象类中，有一个ConcurrentMap>类型的变量map，用来存储消息。第一层是pairtitionID到发送到该partition消息的映射；第二层是VertexID 到发送给该Vertex的消息队列。

《Giraph通信模块分析》：pair存储在ByteArrayVertexIdData中（实际为ByteArrayVertexIdMessages类型）。介绍如下： org.apache.giraph.utils.ByteArrayVertexIdData

功能：把<顶点ID，data> Pair 存储在一个 byte数组中。里面有 ExtendedDataOutput对象用来存储数据。

org.apache.giraph.comm.SendCache用来缓存发送的信息，然后以“Bulk”模式发送。在Giraph中，每个Worker上可以对应多个分区。消息缓存的阈值是以Worker为单位计算，而不是Partition。

SendCache中有ByteArrayVertexIdData[ ] dataCache数组用来存储发送给每个Partition的消息；有int[ ] dataSizes数组用于记录向每个Worker发送的消息大小，若大于MAX_MSG_REQUEST_SIZE（默认为512KB）就把此Worker上的所有Partition缓存的消息发送到给该Worker，同一Worker内消息也是如此缓存；有int[ ] initBufferSizes数组用于记录每个Worker上的每个Partition的初始化ByteArrayVertexIdData中ExtendedDataOutput对象的大小，同一Worker上的所有Partition初始值相同，该值为平均值。记MAX_MSG_REQUEST_SIZE（message request size）值为M， 该Worker上有P个 partitions，ADDTITIONNAL_MSG_REQUEST_SIZE（比平均值大的因子）默认为0.2f，记为A。则每个Partition的初始大小为：M*(1+A) / P .

由前文知道，每个Worker都有一个NettyWorkerClientRequestProcessor用来发送消息。该类中有SendMessageCache对象用来缓存向外发送的信息。NettyWorkerClientRequestProcessor类中的sendMessageRequest(I,M)

方法如下，用于向某个顶点destVertexId发送消息message。

方法解释：首先根据destVertexId得到对应的partitionId和WorkerInfo，然后把消息add到SendMessageCache中，并返回向该顶点所属Worker发送的消息大小workerMessageSize。若该值大于默认值512KB，则把此Worker对应的所有Partition消息从SendMessageCache中删除，把删除的消息赋值给workerMessages，其类型为PairList> ，key为partitionId，value为发送给该partition的消息列表，最后调用doRequest()方法发送信息。doRequest()方法如下：

可以看到在发送消息时，先判断是否在同一Worker上。如果是的话，调用SendWorkerMessagesRequest的doRequest发送消息；否则使用WorkerClient（底层使用Netty）进行消息发送。下面着重讨论同一Worker内的机制。

org.apache.giraph.comm.requests.SendWorkerMessagesRequest类中的doRequest方法如下：

参数为该Worker的ServerData，代码中的partitionVertexData实际为PairList>workerMessages。遍历来添加到ServerData中的incomingMessageStore中。

ByteArrayMessagesPerVertexStore类中的addPartitionMessages()方法如下：

当用户使用了Combiner，incomingMessageStore对应的类型则为OneMessagePerVertexStore，该类为每个顶点只存储一个消息，而非消息队列。 结构如下图：

当添加一条消息时，会把顶点已对应的消息和要添加的消息调用combine()方法进行合并，然后存储在上述结构图中。addPartitionMessages()方法如下：

在ComputeCallable中的call()方法调用computePartition(Partition)计算完所有Partition上的顶点后，调用WorkerClientRequestProcessor.flush()方法把所有剩余的消息发送出去。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。