/*
    * 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);
   }
 }