giraph-core/src/main/java/org/apache/giraph/graph/GlobalStats.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.graph;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.giraph.bsp.checkpoints.CheckpointStatus;
import org.apache.giraph.partition.PartitionStats;
import org.apache.hadoop.io.Writable;

/**
 * Aggregated stats by the master.
 */
public class GlobalStats implements Writable {
  /** All vertices in the application */
  private long vertexCount = 0;
  /** All finished vertices in the last superstep */
  private long finishedVertexCount = 0;
  /** All edges in the last superstep */
  private long edgeCount = 0;
  /** All messages sent in the last superstep */
  private long messageCount = 0;
  /** All message bytes sent in the last superstep */
  private long messageBytesCount = 0;
  /** Whether the computation should be halted */
  private boolean haltComputation = false;
  /** Bytes of data stored to disk in the last superstep */
  private long oocStoreBytesCount = 0;
  /** Bytes of data loaded to disk in the last superstep */
  private long oocLoadBytesCount = 0;
  /** Lowest percentage of graph in memory throughout the execution */
  private int lowestGraphPercentageInMemory = 100;
  /**
   * Master's decision on whether we should checkpoint and
   * what to do next.
   */
  private CheckpointStatus checkpointStatus =
      CheckpointStatus.NONE;

  /**
   * Add the stats of a partition to the global stats.
   *
   * @param partitionStats Partition stats to be added.
   */
  public void addPartitionStats(PartitionStats partitionStats) {
    this.vertexCount += partitionStats.getVertexCount();
    this.finishedVertexCount += partitionStats.getFinishedVertexCount();
    this.edgeCount += partitionStats.getEdgeCount();
  }

  public long getVertexCount() {
    return vertexCount;
  }

  public long getFinishedVertexCount() {
    return finishedVertexCount;
  }

  public long getEdgeCount() {
    return edgeCount;
  }

  public long getMessageCount() {
    return messageCount;
  }

  public long getMessageBytesCount() {
    return messageBytesCount;
  }

  public boolean getHaltComputation() {
    return haltComputation;
  }

  public void setHaltComputation(boolean value) {
    haltComputation = value;
  }

  public long getOocStoreBytesCount() {
    return oocStoreBytesCount;
  }

  public long getOocLoadBytesCount() {
    return oocLoadBytesCount;
  }

  public CheckpointStatus getCheckpointStatus() {
    return checkpointStatus;
  }

  public void setCheckpointStatus(CheckpointStatus checkpointStatus) {
    this.checkpointStatus = checkpointStatus;
  }

  public int getLowestGraphPercentageInMemory() {
    return lowestGraphPercentageInMemory;
  }

  public void setLowestGraphPercentageInMemory(
      int lowestGraphPercentageInMemory) {
    this.lowestGraphPercentageInMemory = lowestGraphPercentageInMemory;
  }

  /**
   * Add bytes loaded to the global stats.
   *
   * @param oocLoadBytesCount number of bytes to be added
   */
  public void addOocLoadBytesCount(long oocLoadBytesCount) {
    this.oocLoadBytesCount += oocLoadBytesCount;
  }

  /**
   * Add bytes stored to the global stats.
   *
   * @param oocStoreBytesCount number of bytes to be added
   */
  public void addOocStoreBytesCount(long oocStoreBytesCount) {
    this.oocStoreBytesCount += oocStoreBytesCount;
  }

  /**
   * Add messages to the global stats.
   *
   * @param messageCount Number of messages to be added.
   */
  public void addMessageCount(long messageCount) {
    this.messageCount += messageCount;
  }

  /**
   * Add messages to the global stats.
   *
   * @param msgBytesCount Number of message bytes to be added.
   */
  public void addMessageBytesCount(long msgBytesCount) {
    this.messageBytesCount += msgBytesCount;
  }

  @Override
  public void readFields(DataInput input) throws IOException {
    vertexCount = input.readLong();
    finishedVertexCount = input.readLong();
    edgeCount = input.readLong();
    messageCount = input.readLong();
    messageBytesCount = input.readLong();
    oocLoadBytesCount = input.readLong();
    oocStoreBytesCount = input.readLong();
    lowestGraphPercentageInMemory = input.readInt();
    haltComputation = input.readBoolean();
    if (input.readBoolean()) {
      checkpointStatus = CheckpointStatus.values()[input.readInt()];
    } else {
      checkpointStatus = null;
    }
  }

  @Override
  public void write(DataOutput output) throws IOException {
    output.writeLong(vertexCount);
    output.writeLong(finishedVertexCount);
    output.writeLong(edgeCount);
    output.writeLong(messageCount);
    output.writeLong(messageBytesCount);
    output.writeLong(oocLoadBytesCount);
    output.writeLong(oocStoreBytesCount);
    output.writeInt(lowestGraphPercentageInMemory);
    output.writeBoolean(haltComputation);
    output.writeBoolean(checkpointStatus != null);
    if (checkpointStatus != null) {
      output.writeInt(checkpointStatus.ordinal());
    }
  }

  @Override
  public String toString() {
    return "(vtx=" + vertexCount + ",finVtx=" +
        finishedVertexCount + ",edges=" + edgeCount + ",msgCount=" +
        messageCount + ",msgBytesCount=" +
          messageBytesCount + ",haltComputation=" + haltComputation +
        ", checkpointStatus=" + checkpointStatus + ')';
  }
}