Java 类com.google.common.util.concurrent.AtomicDoubleArray 实例源码

/**
 * Creates a new {@link UniformReservoir}.
 *
 * @param size the number of samples to keep in the sampling reservoir
 */
public UniformReservoir(int size) {
    this.values = new AtomicDoubleArray(size);
    for (int i = 0; i < values.length(); i++) {
        values.set(i, 0);
    }
    count.set(0);
}

private double computeL1Norm(AtomicDoubleArray a, AtomicDoubleArray b) {
  double ret = 0.0;
  for (int i = 0; i < a.length(); ++i) {
    ret += Math.abs(a.get(i) - b.get(i));
  }
  return ret;
}

private void iterate(double dampingAmount, LongArrayList noOuts, final LongArrayList[] nodePartitions) {
  AtomicDoubleArray nextPR = new AtomicDoubleArray((int) (maxNodeId + 1));

  // First compute how much mass is trapped at the dangling nodes.
  double dangleSum = 0.0;
  LongIterator iter = noOuts.iterator();
  while (iter.hasNext()) {
    dangleSum += prVector.get((int) iter.nextLong());
  }
  dangleSum = dampingFactor * dangleSum / nodeCount;
  final double d = dangleSum;

  // We use a CountDownLatch as a sync barrier to wait for all threads to finish on their
  // respective partitions.
  final CountDownLatch latch = new CountDownLatch(threads);

  // Start all the worker threads over each partition.
  for (int i=0;i<threads; i++ ) {
    new PageRankWorker(i, nextPR, nodePartitions[i], latch, dampingAmount + d).start();
  }
  // Note that an alternative implementation would be to use a CyclicBarrier so we don't need to
  // respawn new threads each time, but for a graph of any size, the cost of respawning new
  // threads is small relative to the cost the actual iterations.

  // Wait for all the threads to finish.
  try {
    latch.await();
  } catch (InterruptedException ex) {
    // Something bad happened, just abort.
    throw new RuntimeException("Error running PageRank!");
  }

  normL1 = computeL1Norm(prVector, nextPR);
  prVector = nextPR;
}

/**
 * Create a decay scheduler.
 * @param numLevels number of priority levels
 * @param ns config prefix, so that we can configure multiple schedulers
 *           in a single instance.
 * @param conf configuration to use.
 */
public DecayRpcScheduler(int numLevels, String ns, Configuration conf) {
  if(numLevels < 1) {
    throw new IllegalArgumentException("Number of Priority Levels must be " +
        "at least 1");
  }
  this.numLevels = numLevels;
  this.namespace = ns;
  this.decayFactor = parseDecayFactor(ns, conf);
  this.decayPeriodMillis = parseDecayPeriodMillis(ns, conf);
  this.identityProvider = this.parseIdentityProvider(ns, conf);
  this.thresholds = parseThresholds(ns, conf, numLevels);
  this.backOffByResponseTimeEnabled = parseBackOffByResponseTimeEnabled(ns,
      conf);
  this.backOffResponseTimeThresholds =
      parseBackOffResponseTimeThreshold(ns, conf, numLevels);

  // Setup response time metrics
  responseTimeTotalInCurrWindow = new AtomicLongArray(numLevels);
  responseTimeCountInCurrWindow = new AtomicLongArray(numLevels);
  responseTimeAvgInLastWindow = new AtomicDoubleArray(numLevels);
  responseTimeCountInLastWindow = new AtomicLongArray(numLevels);

  topUsersCount =
      conf.getInt(DECAYSCHEDULER_METRICS_TOP_USER_COUNT,
          DECAYSCHEDULER_METRICS_TOP_USER_COUNT_DEFAULT);
  Preconditions.checkArgument(topUsersCount > 0,
      "the number of top users for scheduler metrics must be at least 1");

  // Setup delay timer
  Timer timer = new Timer();
  DecayTask task = new DecayTask(this, timer);
  timer.scheduleAtFixedRate(task, decayPeriodMillis, decayPeriodMillis);

  metricsProxy = MetricsProxy.getInstance(ns, numLevels);
  metricsProxy.setDelegate(this);
}

@Test
public void testLesMisGraph() throws Exception {
  OutIndexedPowerLawMultiSegmentDirectedGraph graph =
      new OutIndexedPowerLawMultiSegmentDirectedGraph(1, 1000, 100, 10, 2,
          new IdentityEdgeTypeMask(), new NullStatsReceiver());

  for (int i=0; i<LES_MIS_GRAPH.length; i++) {
    graph.addEdge(LES_MIS_GRAPH[i][0], LES_MIS_GRAPH[i][1], (byte) 0);
  }

  // Spot check the graph to make sure it's been loaded correctly.
  assertEquals(7, graph.getOutDegree(76));
  assertEquals(new LongArrayList(new long[]{64, 65, 66, 63, 62, 48, 58}), new LongArrayList(graph.getOutEdges(76)));

  assertEquals(1, graph.getOutDegree(30));
  assertEquals(new LongArrayList(new long[]{23}), new LongArrayList(graph.getOutEdges(30)));

  assertEquals(4, graph.getOutDegree(11));
  assertEquals(new LongArrayList(new long[]{10, 3, 2, 0}), new LongArrayList(graph.getOutEdges(11)));

  LongOpenHashSet nodes = new LongOpenHashSet();
  long maxNodeId = 0;
  for (int i=0; i<LES_MIS_GRAPH.length; i++) {
    if ( !nodes.contains(LES_MIS_GRAPH[i][0])) nodes.add(LES_MIS_GRAPH[i][0]);
    if ( !nodes.contains(LES_MIS_GRAPH[i][1])) nodes.add(LES_MIS_GRAPH[i][1]);
    if ( LES_MIS_GRAPH[i][0] > maxNodeId ) maxNodeId = LES_MIS_GRAPH[i][0];
    if ( LES_MIS_GRAPH[i][1] > maxNodeId ) maxNodeId = LES_MIS_GRAPH[i][1];
  }

  assertEquals(76, maxNodeId);
  MultiThreadedPageRank pr = new MultiThreadedPageRank(graph, new LongArrayList(nodes), maxNodeId, 0.85, 10, 1e-15, 3);
  int numIterations = pr.run();
  double normL1 = pr.getL1Norm();
  AtomicDoubleArray pagerank = pr.getPageRankVector();
  assertEquals(10, numIterations);
  assertEquals(0.00108, normL1, 10e-4);

  List<Map.Entry<Long, Double>> scores = new ArrayList<>();
  for (int i=0; i<maxNodeId+1; i++) {
    scores.add(new AbstractMap.SimpleEntry<>((long) i, pagerank.get(i)));
  }

  // Sort by score.
  scores.sort((e1, e2) -> e2.getValue() > e1.getValue() ? 1 : e2.getKey().compareTo(e1.getKey()));

  // We're going to verify that the ranking and score are both correct. These rankings have been verified against an
  // external implementation (JUNG).
  assertEquals(11, (long) scores.get(0).getKey());
  assertEquals(0.1088995, scores.get(0).getValue(), 10e-4);
  assertEquals(0, (long) scores.get(1).getKey());
  assertEquals(0.09538347, scores.get(1).getValue(), 10e-4);
  assertEquals(16, (long) scores.get(2).getKey());
  assertEquals(0.05104386, scores.get(2).getValue(), 10e-4);
  assertEquals(23, (long) scores.get(3).getKey());
  assertEquals(0.04389916, scores.get(3).getValue(), 10e-4);
  assertEquals(25, (long) scores.get(4).getKey());
  assertEquals(0.04095956, scores.get(4).getValue(), 10e-4);
  assertEquals(2, (long) scores.get(5).getKey());
  assertEquals(0.03868165, scores.get(5).getValue(), 10e-4);
  assertEquals(24, (long) scores.get(6).getKey());
  assertEquals(0.03617344, scores.get(6).getValue(), 10e-4);
  assertEquals(48, (long) scores.get(7).getKey());
  assertEquals(0.0290502, scores.get(7).getValue(), 10e-4);
  assertEquals(10, (long) scores.get(8).getKey());
  assertEquals(0.02714507, scores.get(8).getValue(), 10e-4);
  assertEquals(3, (long) scores.get(9).getKey());
  assertEquals(0.02714507, scores.get(9).getValue(), 10e-4);

  double totalMass = 0.0;
  for (int i=0; i<maxNodeId+1; i++) {
    totalMass += scores.get(i).getValue();
  }
  // Total mass should still be 1.0.
  assertEquals(1.0, totalMass, 10e-10);
}

/**
 * Creates a PageRank worker thread.
 *
 * @param id      partition id
 * @param nextPR  the PageRank vector to modify
 * @param nodes   the nodes this thread is responsible for
 * @param latch   countdown latch to synchronize all worker threads for an iteration
 * @param mass    PageRank mass to pass along (from dangling nodes and from damping)
 */
public PageRankWorker(int id, AtomicDoubleArray nextPR, LongArrayList nodes,
                      CountDownLatch latch, double mass) {
  this.id = id;
  this.nextPR = nextPR;
  this.nodes = nodes;
  this.latch = latch;
  this.mass = mass;
}

/**
 * Returns the PageRank vector, or null if PageRank has not yet been run.
 *
 * @return the PageRank vector, or null if PageRank has not yet been run
 */
public AtomicDoubleArray getPageRankVector() {
  return prVector;
}