giraph-core/src/main/java/org/apache/giraph/metrics/WorkerSuperstepMetrics.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.giraph.metrics;

import org.apache.giraph.graph.GraphTaskManager;
import org.apache.giraph.ooc.OutOfCoreEngine;
import org.apache.giraph.ooc.OutOfCoreIOCallable;
import org.apache.giraph.worker.BspServiceWorker;
import org.apache.hadoop.io.Writable;

import com.yammer.metrics.core.Gauge;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.PrintStream;
import java.util.concurrent.TimeUnit;

/**
 * Per-superstep metrics for a Worker.
 */
public class WorkerSuperstepMetrics implements Writable {
  /** Total network communication time */
  private LongAndTimeUnit commTimer;
  /** Time for all compute calls to complete */
  private LongAndTimeUnit computeAllTimer;
  /** Time till first message gets flushed */
  private LongAndTimeUnit timeToFirstMsg;
  /** Total superstep time */
  private LongAndTimeUnit superstepTimer;
  /** Time spent waiting for other workers to finish */
  private LongAndTimeUnit waitRequestsTimer;
  /** Time spent doing GC in a superstep */
  private LongAndTimeUnit superstepGCTimer;
  /** Number of bytes loaded from disk to memory in out-of-core mechanism */
  private long bytesLoadedFromDisk;
  /** Number of bytes stored from memory to disk in out-of-core mechanism */
  private long bytesStoredOnDisk;
  /** Percentage of graph kept in memory */
  private double graphPercentageInMemory;

  /**
   * Constructor
   */
  public WorkerSuperstepMetrics() {
    commTimer = new LongAndTimeUnit();
    computeAllTimer = new LongAndTimeUnit();
    timeToFirstMsg = new LongAndTimeUnit();
    superstepTimer = new LongAndTimeUnit();
    waitRequestsTimer = new LongAndTimeUnit();
    superstepGCTimer = new LongAndTimeUnit();
    superstepGCTimer.setTimeUnit(TimeUnit.MILLISECONDS);
    bytesLoadedFromDisk = 0;
    bytesStoredOnDisk = 0;
    graphPercentageInMemory = 100;
  }

  /**
   * Read metric values from global MetricsRegistry.
   *
   * @return this object, for chaining
   */
  public WorkerSuperstepMetrics readFromRegistry() {
    readGiraphTimer(GraphTaskManager.TIMER_COMMUNICATION_TIME, commTimer);
    readGiraphTimer(GraphTaskManager.TIMER_COMPUTE_ALL, computeAllTimer);
    readGiraphTimer(GraphTaskManager.TIMER_TIME_TO_FIRST_MSG, timeToFirstMsg);
    readGiraphTimer(GraphTaskManager.TIMER_SUPERSTEP_TIME, superstepTimer);
    readGiraphTimer(BspServiceWorker.TIMER_WAIT_REQUESTS, waitRequestsTimer);
    SuperstepMetricsRegistry registry = GiraphMetrics.get().perSuperstep();
    superstepGCTimer.setValue(
        registry.getCounter(GraphTaskManager.TIMER_SUPERSTEP_GC_TIME).count());
    bytesLoadedFromDisk =
        registry.getCounter(OutOfCoreIOCallable.BYTES_LOAD_FROM_DISK).count();
    bytesStoredOnDisk =
        registry.getCounter(OutOfCoreIOCallable.BYTES_STORE_TO_DISK).count();
    Gauge<Double> gauge =
        registry.getExistingGauge(OutOfCoreEngine.GRAPH_PERCENTAGE_IN_MEMORY);
    if (gauge != null) {
      graphPercentageInMemory = gauge.value();
    }
    return this;
  }

  /**
   * Read data from GiraphTimer into a LongAndTimeUnit.
   *
   * @param name String name of Gauge to retrieve.
   * @param data LongAndTimeUnit to read data into.
   */
  private void readGiraphTimer(String name, LongAndTimeUnit data) {
    Gauge<Long> gauge = GiraphMetrics.get().perSuperstep().
        getExistingGauge(name);
    if (gauge instanceof GiraphTimer) {
      GiraphTimer giraphTimer = (GiraphTimer) gauge;
      data.setTimeUnit(giraphTimer.getTimeUnit());
      data.setValue(giraphTimer.value());
    } else if (gauge != null) {
      throw new IllegalStateException(name + " is not a GiraphTimer");
    }
  }

  /**
   * Human readable dump of metrics stored here.
   *
   * @param superstep long number of superstep.
   * @param out PrintStream to write to.
   * @return this object, for chaining
   */
  public WorkerSuperstepMetrics print(long superstep, PrintStream out) {
    out.println();
    out.println("--- METRICS: superstep " + superstep + " ---");
    out.println("  superstep time: " + superstepTimer);
    out.println("  compute all partitions: " + computeAllTimer);
    out.println("  time spent in gc: " + superstepGCTimer);
    out.println("  bytes transferred in out-of-core: " +
        (bytesLoadedFromDisk + bytesStoredOnDisk));
    out.println("  network communication time: " + commTimer);
    out.println("  time to first message: " + timeToFirstMsg);
    out.println("  wait on requests time: " + waitRequestsTimer);
    return this;
  }

  /**
   * @return Communication timer
   */
  public long getCommTimer() {
    return commTimer.getValue();
  }

  /**
   * @return Total compute timer
   */
  public long getComputeAllTimer() {
    return computeAllTimer.getValue();
  }

  /**
   * @return timer between start time and first message flushed.
   */
  public long getTimeToFirstMsg() {
    return timeToFirstMsg.getValue();
  }

  /**
   * @return timer for superstep time
   */
  public long getSuperstepTimer() {
    return superstepTimer.getValue();
  }

  /**
   * @return timer waiting for other workers
   */
  public long getWaitRequestsTimer() {
    return waitRequestsTimer.getValue();
  }

  /**
   * @return number of bytes loaded from disk by out-of-core mechanism (if any
   *         is used)
   */
  public long getBytesLoadedFromDisk() {
    return bytesLoadedFromDisk;
  }

  /**
   * @return number of bytes stored on disk by out-of-core mechanism (if any is
   *         used)
   */
  public long getBytesStoredOnDisk() {
    return bytesStoredOnDisk;
  }

  /**
   * @return a rough estimate of percentage of graph in memory
   */
  public double getGraphPercentageInMemory() {
    return graphPercentageInMemory;
  }

  @Override
  public void readFields(DataInput dataInput) throws IOException {
    commTimer.setValue(dataInput.readLong());
    computeAllTimer.setValue(dataInput.readLong());
    timeToFirstMsg.setValue(dataInput.readLong());
    superstepTimer.setValue(dataInput.readLong());
    waitRequestsTimer.setValue(dataInput.readLong());
    bytesLoadedFromDisk = dataInput.readLong();
    bytesStoredOnDisk = dataInput.readLong();
    graphPercentageInMemory = dataInput.readDouble();
  }

  @Override
  public void write(DataOutput dataOutput) throws IOException {
    dataOutput.writeLong(commTimer.getValue());
    dataOutput.writeLong(computeAllTimer.getValue());
    dataOutput.writeLong(timeToFirstMsg.getValue());
    dataOutput.writeLong(superstepTimer.getValue());
    dataOutput.writeLong(waitRequestsTimer.getValue());
    dataOutput.writeLong(bytesLoadedFromDisk);
    dataOutput.writeLong(bytesStoredOnDisk);
    dataOutput.writeDouble(graphPercentageInMemory);
  }
}