/*
    * 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.comm.netty;
  
  import com.yammer.metrics.core.Histogram;
  import com.yammer.metrics.core.Meter;
  import com.yammer.metrics.core.NoOpHistogram;
  import com.yammer.metrics.core.NoOpMeter;
  import io.netty.buffer.ByteBuf;
  import org.apache.giraph.metrics.MeterDesc;
  import org.apache.giraph.metrics.MetricNames;
  import org.apache.giraph.metrics.SuperstepMetricsRegistry;
  import org.apache.giraph.time.SystemTime;
  import org.apache.giraph.time.Time;
  
  import java.text.DecimalFormat;
  import java.util.concurrent.atomic.AtomicLong;
  
  /**
   * Delegate Object to help keep track of the bytes processed and provide some
   * metrics when desired as part of the Netty Channel stack.
   */
  public class ByteCounterDelegate implements ByteCounter {
    /** Megabyte in bytes */
    public static final double MEGABYTE = 1024f * 1024f;
    /** Helper to format the doubles */
    private static final DecimalFormat DOUBLE_FORMAT =
        new DecimalFormat("#######.####");
    /** Class timer */
    private static final Time TIME = SystemTime.get();
    /** Bytes processed during the most recent time interval */
    private final AtomicLong bytesProcessed = new AtomicLong();
    /** Aggregate bytes per superstep */
    private final AtomicLong bytesProcessedPerSuperstep = new AtomicLong();
    /** Total processed requests */
    private final AtomicLong processedRequests = new AtomicLong();
    /** Start time (for bandwidth calculation) */
    private final AtomicLong startMsecs = new AtomicLong();
    /** Last updated msecs for getMetricsWindow */
    private final AtomicLong metricsWindowLastUpdatedMsecs = new AtomicLong();
  
    // Metrics
    /** Meter of requests sent */
    private Meter processedRequestsMeter = NoOpMeter.INSTANCE;
    /** Histogram of bytes sent */
    private Histogram processedBytesHist = NoOpHistogram.INSTANCE;
  
    /** Is it delegate for InBoundByteCounter */
    private final boolean isInbound;
  
    /**
     * Constructor to specify if delegate is created by InBound/ Outbound counter
     *
     * @param isInBound switch to specify if instantiated by inbound counter
     */
    public ByteCounterDelegate(boolean isInBound) {
      this.isInbound = isInBound;
    }
  
    /**
     * Called by Inbound/ Outbound counters to refresh meters on a new superstep
     *
     * @param superstepMetrics superstepmetrics registry
     */
    public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {
      if (isInbound) {
        processedRequestsMeter = superstepMetrics.getMeter(
            MeterDesc.RECEIVED_REQUESTS);
        processedBytesHist = superstepMetrics.getUniformHistogram(
            MetricNames.RECEIVED_BYTES);
      } else {
        processedRequestsMeter = superstepMetrics.getMeter(
            MeterDesc.SENT_REQUESTS);
        processedBytesHist = superstepMetrics.getUniformHistogram(
            MetricNames.SENT_BYTES);
      }
    }
  
    /**
     * Updates properties based on bytes sent / received
     *
     * @param buf ByteBuf received by the counter
     * @return number of readable bytes
    */
   public int byteBookkeeper(ByteBuf buf) {
     int processedBytes = buf.readableBytes();
     bytesProcessed.addAndGet(processedBytes);
     bytesProcessedPerSuperstep.addAndGet(processedBytes);
     processedBytesHist.update(processedBytes);
     processedRequests.incrementAndGet();
     processedRequestsMeter.mark();
     return processedBytes;
   }
 
   /**
    * Reset all the bytes kept track of.
    */
   public void resetBytes() {
     bytesProcessed.set(0);
     processedRequests.set(0);
   }
 
   /**
    * Reset the start msecs.
    */
   public void resetStartMsecs() {
     startMsecs.set(TIME.getMilliseconds());
   }
 
   @Override
   public void resetAll() {
     resetBytes();
     resetStartMsecs();
   }
 
   /**
    * Returns bytes processed per superstep.
    * @return Number of bytes.
    */
   public long getBytesProcessedPerSuperstep() {
     return bytesProcessedPerSuperstep.get();
   }
 
   /**
    * Set bytes processed per superstep to 0.
    */
   public void resetBytesProcessedPerSuperstep() {
     bytesProcessedPerSuperstep.set(0);
   }
 
   public long getBytesProcessed() {
     return bytesProcessed.get();
   }
 
   /**
    * @return Mbytes processed / sec in the current interval
    */
   public double getMbytesPerSecProcessed() {
     return bytesProcessed.get() * 1000f /
         (1 + TIME.getMilliseconds() - startMsecs.get()) / MEGABYTE;
   }
 
   /**
    * Helper method used by getMetrics to create its return string
    * @param mBytesProcessed mbytes processed
    * @param mBytesProcessedPerReq mbytes processed per request
    * @return A string containing all the metrics
    */
   public String getMetricsHelper(double mBytesProcessed,
                                  double mBytesProcessedPerReq) {
     if (isInbound) {
       return "MBytes/sec received = " +
           DOUBLE_FORMAT.format(getMbytesPerSecProcessed()) +
           ", MBytesReceived = " + DOUBLE_FORMAT.format(mBytesProcessed) +
           ", ave received req MBytes = " +
           DOUBLE_FORMAT.format(mBytesProcessedPerReq) +
           ", secs waited = " +
           ((TIME.getMilliseconds() - startMsecs.get()) / 1000f);
     } else {
       return "MBytes/sec sent = " +
           DOUBLE_FORMAT.format(getMbytesPerSecProcessed()) +
           ", MBytesSent = " + DOUBLE_FORMAT.format(mBytesProcessed) +
           ", ave sent req MBytes = " +
           DOUBLE_FORMAT.format(mBytesProcessedPerReq) +
           ", secs waited = " +
           ((TIME.getMilliseconds() - startMsecs.get()) / 1000f);
     }
   }
 
   @Override
   public String getMetrics() {
     double mBytesProcessed = bytesProcessed.get() / MEGABYTE;
     long curProcessedRequests = processedRequests.get();
     double mBytesProcessedPerReq = (curProcessedRequests == 0) ? 0 :
         mBytesProcessed / curProcessedRequests;
 
     return getMetricsHelper(mBytesProcessed, mBytesProcessedPerReq);
   }
 
   @Override
   public String getMetricsWindow(int minMsecsWindow) {
     long lastUpdatedMsecs =  metricsWindowLastUpdatedMsecs.get();
     long curMsecs = TIME.getMilliseconds();
     if (curMsecs - lastUpdatedMsecs > minMsecsWindow) {
       // Make sure that only one thread does this update
       if (metricsWindowLastUpdatedMsecs.compareAndSet(lastUpdatedMsecs,
           curMsecs)) {
         String metrics = getMetrics();
         resetAll();
         return metrics;
       }
     }
     return null;
   }
 }