/*
 * 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.iotdb.db.cost.statistic;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.db.concurrent.ThreadName;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.JMXService;
import org.apache.iotdb.db.service.ServiceType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <p>
 * Measurement is used to record execution time of operations defined in enum class Operation. It
 * can display average time of each operation, and proportion of operation whose execution time fall
 * into time range defined in BUCKET_IN_MS. If you want to change abscissa of histogram, just change
 * the BUCKET_IN_MS array. For recording a operation, you should:
 * 1) add a item in enum class Operation.
 * 2) call <code>startTimeInNano = System.nanoTime()</code> to recode startTime of that operation.
 * 3) call <code>Measurement.INSTANCE.addOperationLatency(operation, startTimeInNano)</code>
 * at the end of that operation;
 *
 * @see Operation
 */
public class Measurement implements MeasurementMBean, IService {
  private static Logger logger = LoggerFactory.getLogger(Measurement.class);

  /**
   * queue for async store time latencies.
   */
  private ConcurrentCircularArray[] operationLatenciesQueue;

  /**
   * size of each queue, this is calculated by memory.
   */
  private final int queueSize;

  /**
   * latencies sum of each operation.
   */
  private long[] operationLatencies;

  /**
   * the num of each operation.
   */
  private long[] operationCnt;

  /**
   * abscissa of histogram.
   */
  private static final int[] BUCKET_IN_MS = {1, 4, 16, 64, 256, 1024, Integer.MAX_VALUE};

  /**
   * length of BUCKET_IN_MS.
   */
  private static final int BUCKET_SIZE = BUCKET_IN_MS.length;

  /**
   * the num of operation that execution time falls into time range of BUCKET_IN_MS. The outer array
   * is each operation, the inner array is each time range in BUCKET_IN_MS.
   */
  private long[][] operationHistogram;

  /**
   * display thread and queue consumer thread.
   */
  private ScheduledExecutorService service;

  /**
   * future task of display thread and queue consumer thread.
   */
  private ScheduledFuture<?> displayFuture;
  private ScheduledFuture<?> consumeFuture;

  /**
   * lock for modifying isEnableStat, displayFuture and consumeFuture.
   */
  private ReentrantLock stateChangeLock = new ReentrantLock();

  public static final Measurement INSTANCE = AsyncMeasurementHolder.MEASUREMENT;

  private boolean isEnableStat;
  private long displayIntervalInMs;
  private Map<String, Boolean> operationSwitch;

  private static final Logger LOGGER = LoggerFactory.getLogger(Measurement.class);
  private final String mbeanName = String
      .format("%s:%s=%s", "org.apache.iotdb.db.cost.statistic", IoTDBConstant.JMX_TYPE,
          getID().getJmxName());

  private Measurement() {
    IoTDBConfig tdbConfig = IoTDBDescriptor.getInstance().getConfig();
    isEnableStat = tdbConfig.isEnablePerformanceStat();
    displayIntervalInMs = tdbConfig.getPerformanceStatDisplayInterval();
    int memoryInKb = tdbConfig.getPerformanceStatMemoryInKB();

    queueSize = memoryInKb * 1000 / Operation.values().length / 8;
    operationLatenciesQueue = new ConcurrentCircularArray[Operation.values().length];
    operationLatencies = new long[Operation.values().length];
    operationCnt = new long[Operation.values().length];
    operationSwitch = new HashMap<>(Operation.values().length);
    for (Operation op : Operation.values()) {
      operationLatenciesQueue[op.ordinal()] = new ConcurrentCircularArray(queueSize);
      operationCnt[op.ordinal()] = 0;
      operationLatencies[op.ordinal()] = 0;
      operationSwitch.put(op.getName(), true);
    }
    operationHistogram = new long[Operation.values().length][BUCKET_SIZE];
    for (Operation operation : Operation.values()) {
      for (int i = 0; i < BUCKET_SIZE; i++) {
        operationHistogram[operation.ordinal()][i] = 0;
      }
    }
    logger.info("start measurement stats module...");
    service = IoTDBThreadPoolFactory.newScheduledThreadPool(
        2, ThreadName.TIME_COST_STATSTIC.getName());
  }

  public boolean addOperationLatency(Operation op, long startTime) {
    if (isEnableStat && operationSwitch.get(op.getName())) {
      return operationLatenciesQueue[op.ordinal()].put((System.currentTimeMillis() - startTime));
    }
    return false;
  }

  @Override
  public void startStatistics() {
    stateChangeLock.lock();
    try {
      isEnableStat = true;
      if (consumeFuture != null && !consumeFuture.isCancelled()) {
        logger.info("The consuming task in measurement stat module is already running...");
      } else {
        consumeFuture = service.scheduleWithFixedDelay(
            new Measurement.DisplayRunnable(), 20, displayIntervalInMs, TimeUnit.MILLISECONDS);
      }
    } catch (Exception e) {
      LOGGER.error("Find error when start performance statistic thread, because {}", e);
    } finally {
      stateChangeLock.unlock();
    }
  }

  @Override
  public void startContinuousPrintStatistics() {
    stateChangeLock.lock();
    try {
      isEnableStat = true;
      if (displayFuture != null && !displayFuture.isCancelled()) {
        logger.info("The display task in measurement stat module is already running...");
      } else {
        displayFuture = service.scheduleWithFixedDelay(
            new Measurement.DisplayRunnable(), 20, displayIntervalInMs, TimeUnit.MILLISECONDS);
      }
    } catch (Exception e) {
      LOGGER.error("Find error when start performance statistic thread, because {}", e);
    } finally {
      stateChangeLock.unlock();
    }
  }

  @Override
  public void startPrintStatisticsOnce() {
    showMeasurements();
  }


  @Override
  public void stopPrintStatistic() {
    stateChangeLock.lock();
    try {
      displayFuture = cancelFuture(displayFuture);
    } catch (Exception e) {
      LOGGER.error("Find error when stop display thread, because {}", e);
    } finally {
      stateChangeLock.unlock();
    }
  }

  @Override
  public void stopStatistic() {
    stateChangeLock.lock();
    try {
      isEnableStat = false;
      displayFuture = cancelFuture(displayFuture);
      consumeFuture = cancelFuture(consumeFuture);
    } catch (Exception e) {
      LOGGER.error("Find error when stop display and consuming threads, because {}", e);
    } finally {
      stateChangeLock.unlock();
    }
  }

  @Override
  public void clearStatisticalState() {
    for (Operation op : Operation.values()) {
      operationLatenciesQueue[op.ordinal()].clear();
      operationCnt[op.ordinal()] = 0;
      operationLatencies[op.ordinal()] = 0;
      for (int i = 0; i < BUCKET_SIZE; i++) {
        operationHistogram[op.ordinal()][i] = 0;
      }
    }
  }

  @Override
  public boolean changeOperationSwitch(String operationName, Boolean operationState) {
    if (operationSwitch.containsKey(operationName)) {
      operationSwitch.put(operationName, operationState);
      return true;
    } else {
      return false;
    }
  }

  /**
   * start service.
   */
  @Override
  public void start() throws StartupException {
    // start display thread and consumer threads.
    logger.info("start the consuming task in the measurement stats module...");
    this.clearStatisticalState();
    if (service.isShutdown()) {
      service = IoTDBThreadPoolFactory.newScheduledThreadPool(
          2, ThreadName.TIME_COST_STATSTIC.getName());
    }
    //we have to check again because someone may change the value.
    isEnableStat = IoTDBDescriptor.getInstance().getConfig().isEnablePerformanceStat();
    if (isEnableStat) {
      consumeFuture = service.schedule(new QueueConsumerThread(), 0, TimeUnit.MILLISECONDS);
    }
    try {
      JMXService.registerMBean(INSTANCE, mbeanName);
    } catch (Exception e) {
      throw new StartupException(this.getID().getName(), e.getMessage());
    }
  }

  /**
   * stop service.
   */
  @Override
  public void stop() {
    logger.info("stop measurement stats module...");
    JMXService.deregisterMBean(mbeanName);
    if (service == null || service.isShutdown()) {
      return;
    }
    service.shutdownNow();
    try {
      consumeFuture = cancelFuture(consumeFuture);
      displayFuture = cancelFuture(displayFuture);
      service.awaitTermination(5, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
      LOGGER.error("Performance statistic service could not be shutdown, {}", e.getMessage());
      // Restore interrupted state...
      Thread.currentThread().interrupt();
    }
  }

  /**
   *
   * @param future
   * @return always return null;
   */
  private ScheduledFuture<?> cancelFuture(ScheduledFuture<?> future) {
    if (future != null) {
      future.cancel(true);
    }
    return null;
  }

  @Override
  public ServiceType getID() {
    return ServiceType.PERFORMANCE_STATISTIC_SERVICE;
  }

  @Override
  public boolean isEnableStat() {
    return isEnableStat;
  }

  @Override
  public long getDisplayIntervalInMs() {
    return displayIntervalInMs;
  }

  @Override
  public void setDisplayIntervalInMs(long displayIntervalInMs) {
    this.displayIntervalInMs = displayIntervalInMs;
  }

  @Override
  public Map<String, Boolean> getOperationSwitch() {
    return operationSwitch;
  }

  private static class AsyncMeasurementHolder {

    private static final Measurement MEASUREMENT = new Measurement();

    private AsyncMeasurementHolder() {
    }
  }

  private void showMeasurements() {
    Date date = new Date();
    LOGGER.info(
        "====================================={} Measurement (ms)======================================",
        date);
    String head = String
        .format("%-45s%-25s%-25s%-25s", "OPERATION", "COUNT", "TOTAL_TIME", "AVG_TIME");
    LOGGER.info(head);
    for (Operation operation : Operation.values()) {
      if (!operationSwitch.get(operation.getName())) {
        continue;
      }
      long cnt = operationCnt[operation.ordinal()];
      long totalInMs = operationLatencies[operation.ordinal()];
      String avg = String.format("%.4f", (totalInMs / (cnt + 1e-9)));
      String item = String
          .format("%-45s%-25s%-25s%-25s", operation.name, cnt + "", totalInMs + "", avg);
      LOGGER.info(item);
    }
    LOGGER.info(
        "==========================================OPERATION HISTOGRAM====================================================");
    StringBuilder histogramHead = new StringBuilder(String.format("%-45s", "OPERATION"));
    for (int i = 0; i < BUCKET_SIZE; i++) {
      histogramHead.append(String.format("%-8s", BUCKET_IN_MS[i] + "ms"));
    }
    if (LOGGER.isInfoEnabled()) {
      LOGGER.info(histogramHead.toString());
    }
    for (Operation operation : Operation.values()) {
      if (!operationSwitch.get(operation.getName())) {
        continue;
      }
      StringBuilder item = new StringBuilder(String.format("%-45s", operation.getName()));
      long cnt = operationCnt[operation.ordinal()];
      for (int i = 0; i < BUCKET_SIZE; i++) {
        String avg = String
            .format("%.2f", (operationHistogram[operation.ordinal()][i] / (cnt + 1e-9) * 100));
        item.append(String.format("%-8s", avg + "%"));
      }
      if (LOGGER.isInfoEnabled()) {
        LOGGER.info(item.toString());
      }
    }

    LOGGER.info(
        "=================================================================================================================");
  }

  class DisplayRunnable implements Runnable {

    @Override
    public void run() {
      showMeasurements();
    }
  }

  class QueueConsumerThread implements Runnable {

    @Override
    public void run() {
      consumer();
    }

    private void consumer() {
      boolean allEmpty;
      while (isEnableStat) {
        allEmpty = true;
        for (Operation op : Operation.values()) {
          if (!operationSwitch.get(op.getName())) {
            continue;
          }
          int idx = op.ordinal();
          ConcurrentCircularArray queue = operationLatenciesQueue[idx];
          if (queue.hasData()) {
            long time = queue.take();
            operationLatencies[idx] += time;
            operationCnt[idx]++;
            operationHistogram[idx][calIndex(time)]++;
            allEmpty = false;
          }
        }
        if (allEmpty) {
          try {
            Thread.sleep(10);
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
          }
        }
      }
    }

    private int calIndex(long x) {
      for (int i = 0; i < BUCKET_SIZE; i++) {
        if (BUCKET_IN_MS[i] >= x) {
          return i;
        }
      }
      return BUCKET_SIZE - 1;
    }
  }

  public long[] getOperationLatencies() {
    return operationLatencies;
  }

  public long[] getOperationCnt() {
    return operationCnt;
  }
}
