Figure operation flow of HAMA BSP Model

Hama-architecture：

Apache-hama Cluster is based on BSP Based on a framework BSPMaster、（ Multiple ） Unrelated GroomServer node controler 、 It can run independently Zookpeer The cluster consists of .BSPMaster use “ fifo ” The principle is right GroomServer monitor 、job Submission processing of 、 Task allocation and record the whole running dynamics ,BSPMaster call BSP Class setup Method 、bsp Methods and cleanup Method pair superstep Control .GroomServer adopt “HeartBeat” towards BSPMaster Send heartbeat message , towards BSPMaster Report current GroomServer Node cluster status . The status information includes the maximum task quantity and available memory capacity of the node cluster .BSPMaster Start according to the heartbeat information BSP Task handle job Divided into... One by one task, And then task Assigned to GroomServer Calculate the node group ,GroomServer start-up BSPPeer perform GroomServer Assigned task.Zookpeer management BSPPeers The barrier synchronization of , Realization BarrierSynchronisation Mechanism .Zookpeer mainly BSPPeer.sync() Method 、enterBarrier() Methods and leaveBarrier() Methods control BSP Barrier synchronization phase of the mission . Transmission of information 、 The receiving is completed in the barrier synchronization stage .

1、job Submission and distribution of ：

job Submit an allocation outline

stay GraphJob Class waitForCompetition() Call in method submit() Method to job Submit to BSPMaster, The main contents submitted are ：VertexClass In the implementation of 、VertexInputReader In the implementation of 、Edge The properties and Vertex Of ID、value etc. .

according to BSPMaster monitor GroomServer The running state of the cluster ,BSPMaster For submitted job According to the number of tasks that can be run GroomServer The cluster of computing nodes assigns tasks . By default GroomServer The number of running tasks is 1, If task The running task quantity of is not 1, be BSPMaster The maximum task amount allocated is the difference between the maximum task amount borne by the cluster and the current task amount . according to BSPMaster The amount of tasks assigned to complete is right job The processed data is divided , After data segmentation BSPMaster control GroomServer Assign data to each BSPPeer in .

HAMA-Graph-setup and bsp Operation diagram

Hama-Graph The operation of the system also follows BSP-HAMA Framework of the , It's all in BSP.setup() Method 、BSP.bsp() Method 、BSP.cleanup() Methods and BSP.clear() Method . use BSP.setup() Methods and BSP.cleanup() Methods are used to execute the start calculation and end calculation, and output the calculation results .superstep The calculation of is mainly in BSP.bsp() Method , therefore BSP.bsp() It mainly controls the core part of the whole parallel computing .BSP.clear() The method is used to clear this superstep And prepare for the next step superstep Calculation .

Graph Operation Overview

2、 Data loading and superstep initialization

Data loading and superstep The initialization of is mainly started by controlling superstep Calculated BSP.setup() Method .

Overview of data loading

The allocated data passes through BSP.setup() Methods GraphJobRunner.loadVertices() Method to load data into memory .Job After publishing, use VertexInputReader.parseVertex() Method to parse the data , After parsing the data, use hashMap Of put(getNumPeers,GraphJobMessage) Method to temporarily store data in hashMap in . obtain hashMap Properties of ( That is, the parsed value value ) use BSPPeer.send(peerName,GraphJobMessage) Method in peers Information transfer between , The main contents to be delivered are peer Address and GraphJobMessage Information . But there is no guarantee that messages are sent and received in the same order , So in the barrier synchronization phase , The same message sent may not arrive BSPPeer On , also BSPPeer There is information transmission between them . Wait until the information transmission is completed in the barrier grid synchronization phase MessageQueue.poll() Method BSPPeer Get the message .

Wait until all the data is loaded into memory superstep The initialization （ The whole program is executed only once ）.Superstep When initializing, the GraphJobRunner.doInitialSuperstep() To perform the . When all BSPPeer The work to be done after receiving the corresponding message is to superstep Initialization of calculation , This work is regarded as the first... After the drama data is loaded supertep Calculation , Although there are VerticeInfo.startSuperstep() Methods and VerticesInfo.finishstartstep() Method runs but does not really superstep Calculation , Because the vertex is still inactive , Just to set BSPPeer Cluster computing threads .

3、 Barrier synchronization

But how does the information work in the barrier synchronization stage ？
 

Barrier grid synchronization method

In the process of synchronization, information transmission mainly uses outgoingMessageManager（ Output information manager ） and localQueue（ Local message queue ）, adopt outgoingMessageManager（ Output information manager ） hold peer Address and GraphJobMessage Package to outgoingBundles( Output package ) in .

All the information is packaged in outgoingBundle The preparation for information transmission is completed . Then all the information is in the barrier synchronization phase （sync()） To summarize . hold peer Address and GraphJobMessage Information from outgoingBundle After taking it out of the , adopt LocalBSPRunner.tansfer() Method store in hashMap The information in is loaded into localQueueForNextIteration（SynchronizedQueue object ） in , And use MessageQueue.addBundle() Method to package .

When information is loaded into localQueue（ Local message queue ） In the after , All the information starts enterBarrier Stage . stay enterBarrier Phases wait by scheduling threads （wait()） Others have not yet entered the barrier synchronization stage . When the last thread that controls the message enters the barrier synchronization phase , A non empty barrier is provided in the structure , Then the current message thread is executed and other message threads are still in the waiting phase , If the last message thread enters the barrier synchronization phase, there is no non empty barrier in the structure , Then each thread enters the preemptive mode , The thread that grabs the execution right will execute the thread first . It is not until all the message threads enter the barrier synchronization phase that they really start to BSPPeer Information transfer between . During the barrier synchronization phase BSPPeer Transmission between is used in AbstractMessageManager.clearOutgoingMessages() Methods localQueueForNextIteration.getMessageQueue() Method uses localQueue The replacement is completed . In the barrier synchronization phase, when all the taskID The associated information for the identification is passed through localQueue Send it out and go to leaveBarrier Stage .

4、superstep Calculation

 superstep Sketch

One superstep Calculation chart

In a superstep During the calculation BSPPeer Only messages can be sent or the last one can be processed superstep Message received in .

Really start superstep The calculation is in doSuperstep(GpprootraphJobMessage,BSPPeer) Implemented in the method of . When BSPPeer Cluster start superstep Before calculation, you need to use class AtomitInteger Activate the vertex to make it active , Traverse all the information and the arrangement order of vertices , With the same ID The iteration information of starts as the first vertex superstep The calculation of . Method of use startSuperstep() perform superstep The beginning of , Next, customize compute() Method implementation , When no vertex is activated or the number of iterations set by the user is reached, it means superstep end .