import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * This function will wait until all partial aggregated values from all
 * workers are ready and aggregated, and return final aggregated values
 * afterwards.
 *
 * @param workerIds All workers in the job apart from the current one
 * @return Iterable through final aggregated values which this worker owns
 */
public Iterable<Map.Entry<String, Writable>>
getMyAggregatorValuesWhenReady(Set<Integer> workerIds) {
  workersBarrier.waitForRequiredPermits(workerIds);
  if (LOG.isDebugEnabled()) {
    LOG.debug("getMyAggregatorValuesWhenReady: Values ready");
  }
  return Iterables.transform(myAggregatorMap.entrySet(),
      new Function<Map.Entry<String, Aggregator<Writable>>,
          Map.Entry<String, Writable>>() {
        @Override
        public Map.Entry<String, Writable> apply(
            Map.Entry<String, Aggregator<Writable>> aggregator) {
          return new AbstractMap.SimpleEntry<String, Writable>(
              aggregator.getKey(),
              aggregator.getValue().getAggregatedValue());
        }
      });
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * This function will wait until all aggregator requests from workers have
 * arrived, and fill the maps for next superstep when ready.
 *
 * @param workerIds All workers in the job apart from the current one
 * @param previousAggregatedValuesMap Map of values from previous
 *                                    superstep to fill out
 * @param currentAggregatorMap Map of aggregators for current superstep to
 *                             fill out. All aggregators in this map will
 *                             be set to initial value.
 */
public void fillNextSuperstepMapsWhenReady(
    Set<Integer> workerIds,
    Map<String, Writable> previousAggregatedValuesMap,
    Map<String, Aggregator<Writable>> currentAggregatorMap) {
  workersBarrier.waitForRequiredPermits(workerIds);
  if (LOG.isDebugEnabled()) {
    LOG.debug("fillNextSuperstepMapsWhenReady: Aggregators ready");
  }
  previousAggregatedValuesMap.clear();
  previousAggregatedValuesMap.putAll(aggregatedValuesMap);
  for (Map.Entry<String, Class<Aggregator<Writable>>> entry :
      aggregatorClassMap.entrySet()) {
    Aggregator<Writable> aggregator =
        currentAggregatorMap.get(entry.getKey());
    if (aggregator == null) {
      currentAggregatorMap.put(entry.getKey(),
          AggregatorUtils.newAggregatorInstance(entry.getValue(), conf));
    } else {
      aggregator.reset();
    }
  }
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public void sendAggregator(String aggregatorName,
    Class<? extends Aggregator> aggregatorClass,
    Writable aggregatedValue) throws IOException {
  WorkerInfo owner =
      AggregatorUtils.getOwner(aggregatorName, service.getWorkerInfoList());
  int currentSize = sendAggregatorCache.addAggregator(owner.getTaskId(),
      aggregatorName, aggregatorClass, aggregatedValue);
  if (currentSize >= maxBytesPerAggregatorRequest) {
    flushAggregatorsToWorker(owner);
  }
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * Register an aggregator which current worker owns. Thread-safe.
 *
 * @param name Name of aggregator
 * @param aggregatorClass Aggregator class
 */
public void registerAggregator(String name,
    Class<Aggregator<Writable>> aggregatorClass) {
  if (LOG.isDebugEnabled() && myAggregatorMap.isEmpty()) {
    LOG.debug("registerAggregator: The first registration after a reset()");
  }
  myAggregatorMap.putIfAbsent(name,
      AggregatorUtils.newAggregatorInstance(aggregatorClass, conf));
  progressable.progress();
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * Get aggregator class from class name, catch all exceptions.
 *
 * @param aggregatorClassName Class nam of aggregator class
 * @return Aggregator class
 */
public static Class<Aggregator<Writable>> getAggregatorClass(String
    aggregatorClassName) {
  try {
    return (Class<Aggregator<Writable>>) Class.forName(aggregatorClassName);
  } catch (ClassNotFoundException e) {
    throw new IllegalStateException("getAggregatorClass: " +
        "ClassNotFoundException for aggregator class " + aggregatorClassName,
        e);
  }
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * Add aggregator to the cache
 *
 * @param taskId Task id of worker which owns the aggregator
 * @param aggregatorName Name of the aggregator
 * @param aggregatorClass Class of the aggregator
 * @param aggregatedValue Value of the aggregator
 * @return Number of bytes in serialized data for this worker
 * @throws IOException
 */
public int addAggregator(Integer taskId, String aggregatorName,
    Class<? extends Aggregator> aggregatorClass,
    Writable aggregatedValue) throws IOException {
  AggregatorOutputStream out = aggregatorMap.get(taskId);
  if (out == null) {
    out = new AggregatorOutputStream();
    aggregatorMap.put(taskId, out);
  }
  return out.addAggregator(aggregatorName, aggregatorClass,
      aggregatedValue);
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * Write aggregator to the stream and increment internal counter
 *
 * @param aggregatorName Name of the aggregator
 * @param aggregatorClass Class of aggregator
 * @param aggregatedValue Value of aggregator
 * @return Number of bytes occupied by the stream
 * @throws IOException
 */
public int addAggregator(String aggregatorName,
    Class<? extends Aggregator> aggregatorClass,
    Writable aggregatedValue) throws IOException {
  incrementCounter();
  dataOutput.writeUTF(aggregatorName);
  dataOutput.writeUTF(aggregatorClass.getName());
  aggregatedValue.write(dataOutput);
  return getSize();
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * Register the class of the aggregator, received by master or worker.
 *
 * @param name            Aggregator name
 * @param aggregatorClass Class of the aggregator
 */
public void registerAggregatorClass(String name,
    Class<Aggregator<Writable>> aggregatorClass) {
  aggregatorClassMap.put(name, aggregatorClass);
  if (!aggregatorTypesMap.containsKey(aggregatorClass)) {
    aggregatorTypesMap.putIfAbsent(aggregatorClass,
        AggregatorUtils.newAggregatorInstance(aggregatorClass, conf));
  }
  progressable.progress();
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public void doRequest(ServerData serverData) {
  DataInput input = getDataInput();
  AllAggregatorServerData aggregatorData = serverData.getAllAggregatorData();
  try {
    int numAggregators = input.readInt();
    for (int i = 0; i < numAggregators; i++) {
      String aggregatorName = input.readUTF();
      String aggregatorClassName = input.readUTF();
      if (aggregatorName.equals(AggregatorUtils.SPECIAL_COUNT_AGGREGATOR)) {
        LongWritable count = new LongWritable(0);
        count.readFields(input);
        aggregatorData.receivedRequestCountFromMaster(count.get(),
            getSenderTaskId());
      } else {
        Class<Aggregator<Writable>> aggregatorClass =
            AggregatorUtils.getAggregatorClass(aggregatorClassName);
        aggregatorData.registerAggregatorClass(aggregatorName,
            aggregatorClass);
        Writable aggregatorValue =
            aggregatorData.createAggregatorInitialValue(aggregatorName);
        aggregatorValue.readFields(input);
        aggregatorData.setAggregatorValue(aggregatorName, aggregatorValue);
        serverData.getOwnerAggregatorData().registerAggregator(
            aggregatorName, aggregatorClass);
      }
    }
  } catch (IOException e) {
    throw new IllegalStateException("doRequest: " +
        "IOException occurred while processing request", e);
  }
  aggregatorData.receivedRequestFromMaster(getData());
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public void doRequest(ServerData serverData) {
  DataInput input = getDataInput();
  AllAggregatorServerData aggregatorData = serverData.getAllAggregatorData();
  try {
    int numAggregators = input.readInt();
    for (int i = 0; i < numAggregators; i++) {
      String aggregatorName = input.readUTF();
      String aggregatorClassName = input.readUTF();
      if (aggregatorName.equals(AggregatorUtils.SPECIAL_COUNT_AGGREGATOR)) {
        LongWritable count = new LongWritable(0);
        count.readFields(input);
        aggregatorData.receivedRequestCountFromWorker(count.get(),
            getSenderTaskId());
      } else {
        Class<Aggregator<Writable>> aggregatorClass =
            AggregatorUtils.getAggregatorClass(aggregatorClassName);
        aggregatorData.registerAggregatorClass(aggregatorName,
            aggregatorClass);
        Writable aggregatorValue =
            aggregatorData.createAggregatorInitialValue(aggregatorName);
        aggregatorValue.readFields(input);
        aggregatorData.setAggregatorValue(aggregatorName, aggregatorValue);
      }
    }
  } catch (IOException e) {
    throw new IllegalStateException("doRequest: " +
        "IOException occurred while processing request", e);
  }
  aggregatorData.receivedRequestFromWorker();
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public final <A extends Writable> boolean registerPersistentAggregator(
    String name,
    Class<? extends Aggregator<A>> aggregatorClass) throws
    InstantiationException, IllegalAccessException {
  return masterAggregatorUsage.registerPersistentAggregator(
      name, aggregatorClass);
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public <A extends Writable> boolean registerAggregator(String name,
    Class<? extends Aggregator<A>> aggregatorClass) throws
    InstantiationException, IllegalAccessException {
  checkAggregatorName(name);
  return registerAggregator(name, aggregatorClass, false) != null;
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public <A extends Writable> boolean registerPersistentAggregator(String name,
    Class<? extends Aggregator<A>> aggregatorClass) throws
    InstantiationException, IllegalAccessException {
  checkAggregatorName(name);
  return registerAggregator(name, aggregatorClass, true) != null;
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * Helper function for registering aggregators.
 *
 * @param name            Name of the aggregator
 * @param aggregatorClass Class of the aggregator
 * @param persistent      Whether aggregator is persistent or not
 * @param <A>             Aggregated value type
 * @return Newly registered aggregator or aggregator which was previously
 *         created with selected name, if any
 */
private <A extends Writable> AggregatorWrapper<A> registerAggregator
(String name, Class<? extends Aggregator<A>> aggregatorClass,
    boolean persistent) throws InstantiationException,
    IllegalAccessException {
  AggregatorWrapper<A> aggregatorWrapper =
      (AggregatorWrapper<A>) aggregatorMap.get(name);
  if (aggregatorWrapper == null) {
    aggregatorWrapper =
        new AggregatorWrapper<A>(aggregatorClass, persistent, conf);
    aggregatorMap.put(name, (AggregatorWrapper<Writable>) aggregatorWrapper);
  }
  return aggregatorWrapper;
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public <A extends Writable> void aggregate(String name, A value) {
  Aggregator<Writable> aggregator = currentAggregatorMap.get(name);
  if (aggregator != null) {
    progressable.progress();
    synchronized (aggregator) {
      aggregator.aggregate(value);
    }
  } else {
    throw new IllegalStateException("aggregate: " +
        AggregatorUtils.getUnregisteredAggregatorMessage(name,
            currentAggregatorMap.size() != 0, conf));
  }
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
/**
 * Constructor
 *
 * Creates new instances of all aggregators from
 * {@link WorkerAggregatorHandler}
 */
public ThreadLocalWorkerAggregatorUsage() {
  threadAggregatorMap = Maps.newHashMapWithExpectedSize(
      WorkerAggregatorHandler.this.currentAggregatorMap.size());
  for (Map.Entry<String, Aggregator<Writable>> entry :
      WorkerAggregatorHandler.this.currentAggregatorMap.entrySet()) {
    threadAggregatorMap.put(entry.getKey(),
        AggregatorUtils.newAggregatorInstance(
            (Class<Aggregator<Writable>>) entry.getValue().getClass(),
            conf));
  }
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public <A extends Writable> void aggregate(String name, A value) {
  Aggregator<Writable> aggregator = threadAggregatorMap.get(name);
  if (aggregator != null) {
    progressable.progress();
    aggregator.aggregate(value);
  } else {
    throw new IllegalStateException("aggregate: " +
        AggregatorUtils.getUnregisteredAggregatorMessage(name,
            threadAggregatorMap.size() != 0, conf));
  }
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public void finishThreadComputation() {
  // Aggregate the values this thread's vertices provided back to
  // WorkerAggregatorHandler
  for (Map.Entry<String, Aggregator<Writable>> entry :
      threadAggregatorMap.entrySet()) {
    WorkerAggregatorHandler.this.aggregate(entry.getKey(),
        entry.getValue().getAggregatedValue());
  }
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public final <A extends Writable> boolean registerAggregator(
  String name, Class<? extends Aggregator<A>> aggregatorClass)
  throws InstantiationException, IllegalAccessException {
  return getGraphState().getGraphTaskManager().getMasterAggregatorUsage().
      registerAggregator(name, aggregatorClass);
}
 

import org.apache.giraph.aggregators.Aggregator; //导入依赖的package包/类
@Override
public final <A extends Writable> boolean registerPersistentAggregator(
    String name,
    Class<? extends Aggregator<A>> aggregatorClass) throws
    InstantiationException, IllegalAccessException {
  return getGraphState().getGraphTaskManager().getMasterAggregatorUsage().
      registerPersistentAggregator(name, aggregatorClass);
}