import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void initialize(GraphState graphState,
  WorkerClientRequestProcessor<LongWritable, ARPVertexValue, NullWritable>
    workerClientRequestProcessor,
  GraphTaskManager<LongWritable, ARPVertexValue, NullWritable>
    graphTaskManager,
  WorkerGlobalCommUsage workerGlobalCommUsage, WorkerContext workerContext) {
  super.initialize(graphState, workerClientRequestProcessor, graphTaskManager,
    workerGlobalCommUsage, workerContext);
  this.k =
    getConf().getInt(NUMBER_OF_PARTITIONS, DEFAULT_NUMBER_OF_PARTITIONS);
  this.stableThreshold = getConf()
    .getInt(NUMBER_OF_STABLE_ITERATIONS, DEFAULT_NUMBER_OF_STABLE_ITERATIONS);
  this.capacityThreshold =
    getConf().getFloat(CAPACITY_THRESHOLD, DEFAULT_CAPACITY_THRESHOLD);
  this.totalPartitionCapacity = getTotalCapacity();
  this.isPartitioned = getConf()
    .getBoolean(ARPTextVertexInputFormat.PARTITIONED_INPUT,
      ARPTextVertexInputFormat.DEFAULT_PARTITIONED_INPUT);
  this.seed = getConf().getLong(SEED, DEFAULT_SEED);
  if (seed != 0) {
    random = new Random(seed);
  } else {
    random = new Random();
  }
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
/**
 * Initialize, called by infrastructure before the superstep starts.
 * Shouldn't be called by user code.
 * 
 * @param graphState
 *            Graph state
 * @param workerClientRequestProcessor
 *            Processor for handling requests
 * @param graphTaskManager
 *            Graph-wide BSP Mapper for this Vertex
 * @param workerAggregatorUsage
 *            Worker aggregator usage
 * @param workerContext
 *            Worker context
 */
public void initialize(GraphState graphState,
		WorkerClientRequestProcessor<I, V, E> workerClientRequestProcessor,
		GraphTaskManager<I, V, E> graphTaskManager,
		WorkerAggregatorUsage workerAggregatorUsage,
		WorkerContext workerContext) {
	this.graphState = graphState;
	this.workerClientRequestProcessor = workerClientRequestProcessor;
	this.graphTaskManager = graphTaskManager;
	this.workerAggregatorUsage = workerAggregatorUsage;
	this.workerContext = workerContext;

	// Request the creation of the injector vertex right at the
	// initialization
	// addVertexRequest(INJECTOR_VERTEX_ID, INJECTOR_VERTEX_VALUE);
	// LOG.info("[PROMETHEUS] Injector Vertex created in initialization!");
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public void initialize(GraphState graphState,
  WorkerClientRequestProcessor<LongWritable, DiffusionVertexValue,
    NullWritable> workerClientRequestProcessor,
  GraphTaskManager<LongWritable, DiffusionVertexValue, NullWritable>
    graphTaskManager,
  WorkerGlobalCommUsage workerGlobalCommUsage, WorkerContext workerContext) {
  this.k = getConf().getInt(NUMBER_OF_CLUSTERS, DEFAULT_NUMBER_OF_CLUSTERS);
  this.secondaryLoadFactor =
    getConf().getDouble(SECONDARY_LOAD_FACTOR, DEFAULT_SECONDARY_LOAD_FACTOR);
  super.initialize(graphState, workerClientRequestProcessor, graphTaskManager,
    workerGlobalCommUsage, workerContext);
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void initialize(GraphState graphState,
  WorkerClientRequestProcessor<LongWritable, LPVertexValue, NullWritable>
    workerClientRequestProcessor,
  GraphTaskManager<LongWritable, LPVertexValue, NullWritable>
    graphTaskManager,
  WorkerGlobalCommUsage workerGlobalCommUsage, WorkerContext workerContext) {
  super.initialize(graphState, workerClientRequestProcessor, graphTaskManager,
    workerGlobalCommUsage, workerContext);
  this.stabilizationRounds = getConf()
    .getLong(STABILISATION_ROUNDS, DEFAULT_NUMBER_OF_STABILIZATION_ROUNDS);
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
/**
 * Initialize, called by infrastructure before the superstep starts.
 * Shouldn't be called by user code.
 *
 * @param graphState Graph state
 * @param workerClientRequestProcessor Processor for handling requests
 * @param graphTaskManager Graph-wide BSP Mapper for this Vertex
 * @param workerAggregatorUsage Worker aggregator usage
 * @param workerContext Worker context
 */
@Override
public void initialize(
    GraphState graphState,
    WorkerClientRequestProcessor<I, V, E> workerClientRequestProcessor,
    GraphTaskManager<I, V, E> graphTaskManager,
    WorkerAggregatorUsage workerAggregatorUsage,
    WorkerContext workerContext) {
  this.graphState = graphState;
  this.workerClientRequestProcessor = workerClientRequestProcessor;
  this.graphTaskManager = graphTaskManager;
  this.workerAggregatorUsage = workerAggregatorUsage;
  this.workerContext = workerContext;
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
/**
 * Create a user worker context
 *
 * @param graphState State of the graph from the worker
 * @return Instantiated user worker context
 */
@SuppressWarnings("rawtypes")
public WorkerContext createWorkerContext(GraphState<I, V, E, M> graphState) {
  WorkerContext workerContext =
      ReflectionUtils.newInstance(getWorkerContextClass(), this);
  workerContext.setGraphState(graphState);
  return workerContext;
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void initialize(GraphState graphState, WorkerClientRequestProcessor<Text, DoubleWritable, Text> workerClientRequestProcessor, GraphTaskManager<Text, DoubleWritable, Text> graphTaskManager, WorkerGlobalCommUsage workerGlobalCommUsage, WorkerContext workerContext) {
    super.initialize(graphState, workerClientRequestProcessor, graphTaskManager, workerGlobalCommUsage, workerContext);
    DGALoggingUtil.setDGALogLevel(this.getConf());
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void initialize(GraphState graphState, WorkerClientRequestProcessor<Text, LouvainNodeState, LongWritable> workerClientRequestProcessor, GraphTaskManager<Text, LouvainNodeState, LongWritable> graphTaskManager, WorkerGlobalCommUsage workerGlobalCommUsage, WorkerContext workerContext) {
    super.initialize(graphState, workerClientRequestProcessor, graphTaskManager, workerGlobalCommUsage, workerContext);
    DGALoggingUtil.setDGALogLevel(this.getConf());
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void initialize(GraphState graphState, WorkerClientRequestProcessor<Text, Text, Text> workerClientRequestProcessor, GraphTaskManager<Text, Text, Text> graphTaskManager, WorkerGlobalCommUsage workerGlobalCommUsage, WorkerContext workerContext) {
    super.initialize(graphState, workerClientRequestProcessor, graphTaskManager, workerGlobalCommUsage, workerContext);
    DGALoggingUtil.setDGALogLevel(getConf());
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void initialize(GraphState graphState, WorkerClientRequestProcessor<Text, VertexData, Text> workerClientRequestProcessor, GraphTaskManager<Text, VertexData, Text> graphTaskManager, WorkerGlobalCommUsage workerGlobalCommUsage, WorkerContext workerContext) {
    super.initialize(graphState, workerClientRequestProcessor, graphTaskManager, workerGlobalCommUsage, workerContext);
    DGALoggingUtil.setDGALogLevel(this.getConf());
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void initialize(GraphState graphState, WorkerClientRequestProcessor<Text, Text, Text> workerClientRequestProcessor, GraphTaskManager<Text, Text, Text> graphTaskManager, WorkerGlobalCommUsage workerGlobalCommUsage, WorkerContext workerContext) {
    super.initialize(graphState, workerClientRequestProcessor, graphTaskManager, workerGlobalCommUsage, workerContext);
    DGALoggingUtil.setDGALogLevel(this.getConf());
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
public final WorkerContext getWorkerContext() {
  return serviceWorker.getWorkerContext();
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void preSuperstep(WorkerContext workerContext) {
  counter =
      ((TestPartitionContextWorkerContext) workerContext).superstepCounter;
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void postSuperstep(WorkerContext workerContext) {
  ((TestPartitionContextWorkerContext) workerContext).totalCounter +=
      counter;
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void preSuperstep(WorkerContext workerContext) {
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
@Override
public void postSuperstep(WorkerContext workerContext) {
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
/**
 * Compute a single partition
 *
 * @param partition Partition to compute
 * @return Partition stats for this computed partition
 */
private PartitionStats computePartition(Partition<I, V, E, M> partition)
  throws IOException, InterruptedException {
  PartitionStats partitionStats =
      new PartitionStats(partition.getId(), 0, 0, 0, 0);
  // Make sure this is thread-safe across runs
  synchronized (partition) {
    // Prepare Partition context
    WorkerContext workerContext =
        graphState.getGraphTaskManager().getWorkerContext();
    PartitionContext partitionContext = partition.getPartitionContext();
    synchronized (workerContext) {
      partitionContext.preSuperstep(workerContext);
    }
    graphState.setPartitionContext(partition.getPartitionContext());

    for (Vertex<I, V, E, M> vertex : partition) {
      // Make sure every vertex has this thread's
      // graphState before computing
      vertex.setGraphState(graphState);
      Iterable<M> messages = messageStore.getVertexMessages(vertex.getId());
      if (vertex.isHalted() && !Iterables.isEmpty(messages)) {
        vertex.wakeUp();
      }
      if (!vertex.isHalted()) {
        context.progress();
        TimerContext computeOneTimerContext = computeOneTimer.time();
        try {
          vertex.compute(messages);
        } finally {
          computeOneTimerContext.stop();
        }
        // Need to unwrap the mutated edges (possibly)
        vertex.unwrapMutableEdges();
        // Write vertex to superstep output (no-op if it is not used)
        vertexWriter.writeVertex(vertex);
        // Need to save the vertex changes (possibly)
        partition.saveVertex(vertex);
      }
      //################## add by fidel
      else{
      	if(GiraphConstants.MY_TEST.get(configuration))
      		TLOG.info("ht");
      }
      //##################
      if (vertex.isHalted()) {
        partitionStats.incrFinishedVertexCount();
      }
      // Remove the messages now that the vertex has finished computation
      messageStore.clearVertexMessages(vertex.getId());

      // Add statistics for this vertex
      partitionStats.incrVertexCount();
      partitionStats.addEdgeCount(vertex.getNumEdges());
    }

    messageStore.clearPartition(partition.getId());

    synchronized (workerContext) {
      partitionContext.postSuperstep(workerContext);
    }
  }
  return partitionStats;
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
/**
 * Get the worker context
 * 
 * @param <W>
 *            WorkerContext class
 * @return WorkerContext context
 */
@SuppressWarnings("unchecked")
public <W extends WorkerContext> W getWorkerContext() {
	return (W) workerContext;
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
/**
 * Get the worker context
 *
 * @param <W> WorkerContext class
 * @return WorkerContext context
 */
@SuppressWarnings("unchecked")
public <W extends WorkerContext> W getWorkerContext() {
  return (W) giraphCompute.getWorkerContext();
}
 

import org.apache.giraph.worker.WorkerContext; //导入依赖的package包/类
/**
 * Get the worker context.
 *
 * @return worker's WorkerContext
 */
WorkerContext getWorkerContext();