/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Portions Copyright 2006-2007 Sun Microsystems, Inc.
 */
package org.opends.server.backends.jeb;
import org.opends.messages.Message;

import org.opends.server.types.*;

import java.util.ArrayList;
import java.util.TimerTask;
import java.util.Timer;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.ReentrantLock;

import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.StatsConfig;
import com.sleepycat.je.EnvironmentStats;

import static org.opends.server.loggers.ErrorLogger.logError;
import static org.opends.server.loggers.debug.DebugLogger.*;
import org.opends.server.loggers.debug.DebugTracer;
import org.opends.server.core.DirectoryServer;
import static org.opends.messages.JebMessages.
    ERR_JEB_ATTRIBUTE_INDEX_NOT_CONFIGURED;
import static org.opends.messages.JebMessages.
    INFO_JEB_REBUILD_PROGRESS_REPORT;
import static org.opends.messages.JebMessages.
    INFO_JEB_REBUILD_FINAL_STATUS;
import static org.opends.messages.JebMessages.
    INFO_JEB_REBUILD_CACHE_AND_MEMORY_REPORT;
import static org.opends.messages.JebMessages.
    ERR_JEB_REBUILD_INDEX_CONFLICT;
import static org.opends.messages.JebMessages.
    INFO_JEB_REBUILD_START;
import static org.opends.messages.JebMessages.
    ERR_JEB_VLV_INDEX_NOT_CONFIGURED;
/**
 * Runs a index rebuild process on the backend. Each index selected for rebuild
 * will be done from scratch by first clearing out the database for that index.
 * Different threads will be used to rebuild each index.
 * The rebuild process can run concurrently with the backend online and
 * performing write and read operations. However, during the rebuild process,
 * other reader and writer activeThreads might notice inconsistencies in index
 * databases being rebuilt. They can safely ignore these inconsistencies as long
 * as a rebuild is in progress.
 */
public class RebuildJob
{
  /**
   * The tracer object for the debug logger.
   */
  private static final DebugTracer TRACER = getTracer();

  /**
   * The rebuild configuraiton.
   */
  private RebuildConfig rebuildConfig;

  /**
   * The root container used for the verify job.
   */
  private RootContainer rootContainer;

  /**
   * The number of milliseconds between job progress reports.
   */
  private long progressInterval = 10000;

  /**
   * The waiting rebuild threads created to process the rebuild.
   */
  private CopyOnWriteArrayList<IndexRebuildThread> waitingThreads =
      new CopyOnWriteArrayList<IndexRebuildThread>();

  /**
   * The active rebuild threads created to process the rebuild.
   */
  private CopyOnWriteArrayList<IndexRebuildThread> activeThreads =
      new CopyOnWriteArrayList<IndexRebuildThread>();

  /**
   * The completed rebuild threads used to process the rebuild.
   */
  private CopyOnWriteArrayList<IndexRebuildThread> completedThreads =
      new CopyOnWriteArrayList<IndexRebuildThread>();

  /**
   * Rebuild jobs currently running.
   */
  private static CopyOnWriteArrayList<RebuildJob> rebuildJobs =
      new CopyOnWriteArrayList<RebuildJob>();

  /**
   * A mutex that will be used to provide threadsafe access to methods changing
   * the set of currently running rebuild jobs.
   */
  private static ReentrantLock jobsMutex = new ReentrantLock();

  /**
   * This class reports progress of the rebuild job at fixed intervals.
   */
  class ProgressTask extends TimerTask
  {
    /**
     * The number of records that had been processed at the time of the
     * previous progress report.
     */
    private long previousProcessed = 0;

    /**
     * The time in milliseconds of the previous progress report.
     */
    private long previousTime;

    /**
     * The environment statistics at the time of the previous report.
     */
    private EnvironmentStats prevEnvStats;

    /**
     * The number of bytes in a megabyte.
     * Note that 1024*1024 bytes may eventually become known as a mebibyte(MiB).
     */
    private static final int bytesPerMegabyte = 1024*1024;
   /**
     * Create a new verify progress task.
     * @throws DatabaseException An error occurred while accessing the JE
     * database.
     */
    public ProgressTask() throws DatabaseException
    {
      previousTime = System.currentTimeMillis();
      prevEnvStats =
          rootContainer.getEnvironmentStats(new StatsConfig());
    }

    /**
     * The action to be performed by this timer task.
     */
    public void run()
    {
      long latestTime = System.currentTimeMillis();
      long deltaTime = latestTime - previousTime;

      if (deltaTime == 0)
      {
        return;
      }

      long totalEntries = 0;
      long latestProcessed = 0;

      ArrayList<IndexRebuildThread> allThreads =
          new ArrayList<IndexRebuildThread>(waitingThreads);
      allThreads.addAll(activeThreads);
      allThreads.addAll(completedThreads);

      for(IndexRebuildThread thread : allThreads)
      {
        try
        {
          totalEntries += thread.getTotalEntries();
          latestProcessed += thread.getProcessedEntries();

          if(debugEnabled())
          {
            TRACER.debugVerbose("Rebuild thread %s stats: total %d " +
                "processed %d rebuilt %d duplicated %d skipped %d",
                         thread.getTotalEntries(), thread.getProcessedEntries(),
                         thread.getRebuiltEntries(),
                         thread.getDuplicatedEntries(),
                         thread.getSkippedEntries());
          }
        }
        catch(Exception e)
        {
          if(debugEnabled())
          {
            TRACER.debugCaught(DebugLogLevel.ERROR, e);
          }
        }
      }

      long deltaCount = (latestProcessed - previousProcessed);
      float rate = 1000f*deltaCount / deltaTime;
      float completed = 0;
      if(totalEntries > 0)
      {
        completed = 100f*latestProcessed / totalEntries;
      }

      Message message = INFO_JEB_REBUILD_PROGRESS_REPORT.get(
          completed, latestProcessed, totalEntries, rate);
      logError(message);

      try
      {
        Runtime runtime = Runtime.getRuntime();
        long freeMemory = runtime.freeMemory() / bytesPerMegabyte;

        EnvironmentStats envStats =
            rootContainer.getEnvironmentStats(new StatsConfig());
        long nCacheMiss =
             envStats.getNCacheMiss() - prevEnvStats.getNCacheMiss();

        float cacheMissRate = 0;
        if (deltaCount > 0)
        {
          cacheMissRate = nCacheMiss/(float)deltaCount;
        }

        message = INFO_JEB_REBUILD_CACHE_AND_MEMORY_REPORT.get(
            freeMemory, cacheMissRate);
        logError(message);

        prevEnvStats = envStats;
      }
      catch (DatabaseException e)
      {
        if (debugEnabled())
        {
          TRACER.debugCaught(DebugLogLevel.ERROR, e);
        }
      }


      previousProcessed = latestProcessed;
      previousTime = latestTime;
    }
  }

  /**
   * Construct a new rebuild job.
   *
   * @param rebuildConfig The configuration to use for this rebuild job.
   */
  public RebuildJob(RebuildConfig rebuildConfig)
  {
    this.rebuildConfig = rebuildConfig;
  }

  private static void addJob(RebuildJob job)
      throws DatabaseException, JebException
  {
    //Make sure there are no running rebuild jobs
    jobsMutex.lock();

    try
    {
      for(RebuildJob otherJob : rebuildJobs)
      {
        String conflictIndex =
            job.rebuildConfig.checkConflicts(otherJob.rebuildConfig);
        if(conflictIndex != null)
        {
          if(debugEnabled())
          {
            TRACER.debugError("Conflit detected. This job config: %s, " +
                "That job config: %s.",
                              job.rebuildConfig, otherJob.rebuildConfig);
          }

          Message msg = ERR_JEB_REBUILD_INDEX_CONFLICT.get(conflictIndex);
          throw new JebException(msg);
        }
      }

      //No conflicts are found. Add the job to the list of currently running
      // jobs.
      rebuildJobs.add(job);
    }
    finally
    {
      jobsMutex.unlock();
    }
  }

  private static void removeJob(RebuildJob job)
  {
    jobsMutex.lock();

    rebuildJobs.remove(job);

    jobsMutex.unlock();
  }

  /**
   * Initiate the rebuild process on a backend.
   *
   * @param rootContainer The root container to rebuild in.
   * @throws DirectoryException If an error occurs during the rebuild process.
   * @throws DatabaseException If a JE database error occurs during the rebuild
   *                           process.
   * @throws JebException If a JE database error occurs during the rebuild
   *                      process.
   */
  public void rebuildBackend(RootContainer rootContainer)
      throws DirectoryException, DatabaseException, JebException
  {
    //TODO: Add check for only performing internal indexType rebuilds when
    // backend is offline.

    addJob(this);

    try
    {
      this.rootContainer = rootContainer;
      EntryContainer entryContainer =
          rootContainer.getEntryContainer(rebuildConfig.getBaseDN());

      ArrayList<String> rebuildList = rebuildConfig.getRebuildList();

      if(!rebuildList.isEmpty())
      {

        for (String index : rebuildList)
        {
          IndexRebuildThread rebuildThread;
          String lowerName = index.toLowerCase();
          if (lowerName.equals("dn2id"))
          {
            rebuildThread = new IndexRebuildThread(entryContainer,
                                            IndexRebuildThread.IndexType.DN2ID);
          }
          else if (lowerName.equals("dn2uri"))
          {
            rebuildThread = new IndexRebuildThread(entryContainer,
                                           IndexRebuildThread.IndexType.DN2URI);
          }
          else if (lowerName.equals("id2children"))
          {
            rebuildThread = new IndexRebuildThread(entryContainer,
                                      IndexRebuildThread.IndexType.ID2CHILDREN);
          }
          else if (lowerName.equals("id2subtree"))
          {
            rebuildThread = new IndexRebuildThread(entryContainer,
                                       IndexRebuildThread.IndexType.ID2SUBTREE);
          }
          else if (lowerName.startsWith("vlv."))
          {
            if(lowerName.length() < 5)
            {
              Message msg = ERR_JEB_VLV_INDEX_NOT_CONFIGURED.get(lowerName);
              throw new JebException(msg);
            }

            VLVIndex vlvIndex =
                entryContainer.getVLVIndex(lowerName.substring(4));
            if(vlvIndex == null)
            {
              Message msg =
                  ERR_JEB_VLV_INDEX_NOT_CONFIGURED.get(lowerName.substring(4));
              throw new JebException(msg);
            }

            rebuildThread = new IndexRebuildThread(entryContainer, vlvIndex);
          }
          else
          {
            String[] attrIndexParts = lowerName.split("\\.");
            if(attrIndexParts.length <= 0)
            {
              Message msg = ERR_JEB_ATTRIBUTE_INDEX_NOT_CONFIGURED.get(index);
              throw new JebException(msg);
            }

            AttributeType attrType =
                DirectoryServer.getAttributeType(attrIndexParts[0]);

            if (attrType == null)
            {
              Message msg = ERR_JEB_ATTRIBUTE_INDEX_NOT_CONFIGURED.get(index);
              throw new JebException(msg);
            }
            AttributeIndex attrIndex =
                entryContainer.getAttributeIndex(attrType);
            if (attrIndex == null)
            {
              Message msg = ERR_JEB_ATTRIBUTE_INDEX_NOT_CONFIGURED.get(index);
              throw new JebException(msg);
            }

            if(attrIndexParts.length > 1)
            {
              Index partialAttrIndex = null;
              if(attrIndexParts[1].equals("presence"))
              {
                partialAttrIndex = attrIndex.presenceIndex;
              }
              else if(attrIndexParts[1].equals("equality"))
              {
                partialAttrIndex = attrIndex.equalityIndex;
              }
              else if(attrIndexParts[1].equals("substring"))
              {
                partialAttrIndex = attrIndex.substringIndex;
              }
              else if(attrIndexParts[1].equals("ordering"))
              {
                partialAttrIndex = attrIndex.orderingIndex;
              }
              else if(attrIndexParts[1].equals("approximate"))
              {
                partialAttrIndex = attrIndex.approximateIndex;
              }

              if(partialAttrIndex == null)
              {
                Message msg = ERR_JEB_ATTRIBUTE_INDEX_NOT_CONFIGURED.get(index);
                throw new JebException(msg);
              }

              rebuildThread =
                  new IndexRebuildThread(entryContainer, partialAttrIndex);
            }
            else
            {
              rebuildThread = new IndexRebuildThread(entryContainer,
                                                     attrIndex);
            }
          }

          waitingThreads.add(rebuildThread);

          if(debugEnabled())
          {
            TRACER.debugInfo("Created rebuild thread %s",
                             rebuildThread.getName());
          }
        }

        //Log a start message.
        long totalToProcess = 0;

        for(IndexRebuildThread thread : waitingThreads)
        {
          totalToProcess += thread.getTotalEntries();
        }

        StringBuilder sb = new StringBuilder();
        for(String index : rebuildList)
        {
          if(sb.length() > 0)
          {
            sb.append(", ");
          }
          sb.append(index);
        }
        Message message =
            INFO_JEB_REBUILD_START.get(sb.toString(), totalToProcess);
        logError(message);

        // Make a note of the time we started.
        long startTime = System.currentTimeMillis();

        // Start a timer for the progress report.
        Timer timer = new Timer();
        TimerTask progressTask = new ProgressTask();
        timer.scheduleAtFixedRate(progressTask, progressInterval,
                                  progressInterval);

        entryContainer.exclusiveLock.lock();
        try
        {
          for(IndexRebuildThread thread : waitingThreads)
          {
            thread.clearDatabase();
          }
        }
        finally
        {
          if(!rebuildConfig.includesSystemIndex())
          {
            entryContainer.exclusiveLock.unlock();
          }
        }


        if(!rebuildConfig.includesSystemIndex())
        {
          entryContainer.sharedLock.lock();
        }
        try
        {
          while(!waitingThreads.isEmpty())
          {
            dispatchThreads();
            joinThreads();
          }
        }
        finally
        {
          timer.cancel();
          if(rebuildConfig.includesSystemIndex())
          {
            entryContainer.exclusiveLock.unlock();
          }
          else
          {
            entryContainer.sharedLock.unlock();
          }
        }

        long totalProcessed = 0;
        long totalRebuilt = 0;
        long totalDuplicated = 0;
        long totalSkipped = 0;

        for(IndexRebuildThread thread : completedThreads)
        {
          totalProcessed += thread.getProcessedEntries();
          totalRebuilt += thread.getRebuiltEntries();
          totalDuplicated += thread.getDuplicatedEntries();
          totalSkipped += thread.getSkippedEntries();
        }

        long finishTime = System.currentTimeMillis();
        long totalTime = (finishTime - startTime);

        float rate = 0;
        if (totalTime > 0)
        {
          rate = 1000f*totalProcessed / totalTime;
        }

        message = INFO_JEB_REBUILD_FINAL_STATUS.get(
            totalProcessed, totalTime/1000, rate);
        logError(message);

        if(debugEnabled())
        {
          TRACER.debugInfo("Detailed overall rebuild job stats: rebuilt %d, " +
              "duplicated %d, skipped %d",
                    totalRebuilt, totalDuplicated, totalSkipped);
        }
      }
    }
    finally
    {
      removeJob(this);
    }

  }

  /**
   * Dispatch a set of threads based on their dependency and ordering.
   */
  private void dispatchThreads() throws DatabaseException
  {
    for(IndexRebuildThread t : waitingThreads)
    {
      boolean start = true;

      //Check to see if we have exceeded the max number of threads to use at
      //one time.
      if(rebuildConfig.getMaxRebuildThreads() > 0 &&
          activeThreads.size() > rebuildConfig.getMaxRebuildThreads())
      {
        if(debugEnabled())
        {
          TRACER.debugInfo("Delaying the start of thread %s because " +
              "the max number of rebuild threads has been reached.");
        }
        start = false;
      }

      /**
       * We may need to start the threads in stages since the rebuild process
       * of some index types (id2children, id2subtree) depends on another
       * index being rebuilt to be completed first.
       */
      if(t.getIndexType() == IndexRebuildThread.IndexType.ID2CHILDREN ||
          t.getIndexType() == IndexRebuildThread.IndexType.ID2SUBTREE)
      {
        //Check to see if we have any waiting threads that needs to go
        //first
        for(IndexRebuildThread t2 : waitingThreads)
        {
          if(t2.getIndexType() == IndexRebuildThread.IndexType.DN2ID ||
              t2.getIndexType() == IndexRebuildThread.IndexType.DN2URI)
          {
            //We gotta wait for these to start before running the
            //rebuild on ID2CHILDREN or ID2SUBTREE

            if(debugEnabled())
            {
              TRACER.debugInfo("Delaying the start of thread %s because " +
                  "it depends on another index rebuilt to " +
                  "go first.", t.getName());
            }
            start = false;
            break;
          }
        }

        //Check to see if we have any active threads that needs to
        //finish first
        for(IndexRebuildThread t3 : activeThreads)
        {
          if(t3.getIndexType() == IndexRebuildThread.IndexType.DN2ID ||
              t3.getIndexType() == IndexRebuildThread.IndexType.DN2URI)
          {
            //We gotta wait for these to start before running the
            //rebuild on ID2CHILDREN or ID2SUBTREE

            if(debugEnabled())
            {
              TRACER.debugInfo("Delaying the start of thread %s because " +
                  "it depends on another index being rebuilt to " +
                  "finish.", t.getName());
            }
            start = false;
            break;
          }
        }
      }

      if(start)
      {
        if(debugEnabled())
        {
          TRACER.debugInfo("Starting rebuild thread %s.", t.getName());
        }
        waitingThreads.remove(t);
        activeThreads.add(t);
        t.start();
      }
    }
  }

  /**
   * Wait for all worker activeThreads to exit.
   */
  private void joinThreads()
  {
    for (IndexRebuildThread t : activeThreads)
    {
      try
      {
        t.join();

        if(debugEnabled())
        {
          TRACER.debugInfo("Rebuild thread %s finished.", t.getName());
        }
        activeThreads.remove(t);
        completedThreads.add(t);
      }
      catch (InterruptedException ie)
      {
        // No action needed?
      }
    }
  }
}