/*
 * 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 2008 Sun Microsystems, Inc.
 */

package org.opends.server.extensions;
import org.opends.messages.Message;

import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import org.opends.server.api.Backend;
import org.opends.server.api.DirectoryThread;
import org.opends.server.api.ServerShutdownListener;
import org.opends.server.core.DirectoryServer;
import org.opends.server.types.DN;
import org.opends.server.types.DebugLogLevel;
import org.opends.server.types.DirectoryException;
import org.opends.server.loggers.debug.DebugTracer;

import org.opends.server.types.LockManager;
import static org.opends.server.util.StaticUtils.*;
import static org.opends.server.loggers.debug.DebugLogger.*;
import static org.opends.server.loggers.ErrorLogger.logError;
import static org.opends.messages.ExtensionMessages.*;

/**
 * This class defines a utility that will be used to pre-load the Directory
 * Server entry cache.  Pre-loader is multi-threaded and consist of the
 * following threads:
 *
 * - The Arbiter thread which monitors overall pre-load progress and manages
 *   pre-load worker threads by adding or removing them as deemed necessary.
 *
 * - The Collector thread which collects all entry DNs stored within every
 *   configured and active backend to a shared object workers consume from.
 *
 * - Worker threads which are responsible for monitoring the collector feed
 *   and requesting the actual entries for retrieval and in cache storage.
 *
 * This implementation is entry cache and backend independent and can be
 * used to pre-load from any backend to any entry cache as long as both
 * are capable of initiating and sustaining such pre-load activity.
 *
 * This implementation is fully synchronized and safe to use with the server
 * online and pre-load activities going in parallel with server operations.
 *
 * This implementation is self-adjusting to any system workload and does not
 * require any configuration parameters to optimize for initial system
 * resources availability and/or any subsequent fluctuations.
 */
public class EntryCachePreloader
  extends DirectoryThread
  implements ServerShutdownListener
{
  /**
   * The tracer object for the debug logger.
   */
  private static final DebugTracer TRACER = getTracer();

  /**
   * Interrupt flag for the arbiter to terminate worker threads.
   */
  private AtomicBoolean interruptFlag = new AtomicBoolean(false);

  /**
   * Processed entries counter.
   */
  private AtomicLong processedEntries = new AtomicLong(0);

  /**
   * Progress report resolution.
   */
  private static final long progressInterval = 5000;

  /**
   * Default arbiter resolution time.
   */
  public static final long
    PRELOAD_ARBITER_DEFAULT_SLEEP_TIME = 1000;

  /**
   * Effective arbiter resolution time.
   */
  private static long arbiterSleepTime;

  /**
   * Pre-load arbiter thread name.
   */
  private String preloadArbiterThreadName;

  /**
   * Pre-load arbiter thread.
   */
  private Thread preloadArbiterThread;

  /**
   * Worker threads.
   */
  private List<Thread> preloadThreads =
    Collections.synchronizedList(
    new LinkedList<Thread>());

  /**
   * DN Collector thread.
   */
  private EntryCacheDNCollector dnCollector =
    new EntryCacheDNCollector();

  /**
   * This queue is for workers to take from.
   */
  private LinkedBlockingQueue<DN> dnQueue =
      new LinkedBlockingQueue<DN>();

  /**
   * The number of bytes in a megabyte.
   */
  private static final int bytesPerMegabyte = 1024*1024;

  /**
   * Default constructor.
   */
  public EntryCachePreloader() {
    super("Entry Cache Preload Arbiter");
    preloadArbiterThreadName = getName();
    DirectoryServer.registerShutdownListener(this);
    // This should not be exposed as configuration
    // parameter and is only useful for testing.
    arbiterSleepTime = Long.getLong(
      "org.opends.server.entrycache.preload.sleep",
      PRELOAD_ARBITER_DEFAULT_SLEEP_TIME);
  }

  /**
   * The Arbiter thread.
   */
  @Override
  public void run() {
    preloadArbiterThread = Thread.currentThread();
    logError(NOTE_CACHE_PRELOAD_PROGRESS_START.get());
    // Start DN collector thread first.
    dnCollector.start();
    // Kick off a single worker.
    EntryCachePreloadWorker singleWorkerThread =
      new EntryCachePreloadWorker();
    singleWorkerThread.start();
    preloadThreads.add(singleWorkerThread);
    // Progress report timer task.
    Timer timer = new Timer();
    TimerTask progressTask = new TimerTask() {
      // Persistent state restore progress report.
      public void run() {
        if (processedEntries.get() > 0) {
          long freeMemory =
            Runtime.getRuntime().freeMemory() / bytesPerMegabyte;
          Message message = NOTE_CACHE_PRELOAD_PROGRESS_REPORT.get(
            processedEntries.get(), freeMemory);
          logError(message);
        }
      }
    };
    timer.scheduleAtFixedRate(progressTask, progressInterval,
      progressInterval);
    // Cycle to monitor progress and adjust workers.
    long processedEntriesDeltaLow  = 0;
    long processedEntriesDeltaHigh = 0;
    long lastKnownProcessedEntries = 0;
    try {
      while (!dnQueue.isEmpty() || dnCollector.isAlive()) {
        long processedEntriesCycle = processedEntries.get();
        long processedEntriesDelta =
          processedEntriesCycle - lastKnownProcessedEntries;
        lastKnownProcessedEntries = processedEntriesCycle;
        // Spawn another worker if scaling up.
        if (processedEntriesDelta > processedEntriesDeltaHigh) {
          processedEntriesDeltaLow = processedEntriesDeltaHigh;
          processedEntriesDeltaHigh = processedEntriesDelta;
          EntryCachePreloadWorker workerThread =
            new EntryCachePreloadWorker();
          workerThread.start();
          preloadThreads.add(workerThread);
        }
        // Interrupt random worker if scaling down.
        if (processedEntriesDelta < processedEntriesDeltaLow) {
          processedEntriesDeltaHigh = processedEntriesDeltaLow;
          processedEntriesDeltaLow = processedEntriesDelta;
          // Leave at least one worker to progress.
          if (preloadThreads.size() > 1) {
            interruptFlag.set(true);
          }
        }
        Thread.sleep(arbiterSleepTime);
      }
      // Join the collector.
      dnCollector.join();
      // Join all spawned workers.
      for (Thread workerThread : preloadThreads) {
        workerThread.join();
      }
      // Cancel progress report task and report done.
      timer.cancel();
      Message message = NOTE_CACHE_PRELOAD_PROGRESS_DONE.get(
        processedEntries.get());
      logError(message);
    } catch (InterruptedException ex) {
      if (debugEnabled()) {
        TRACER.debugCaught(DebugLogLevel.ERROR, ex);
      }
      // Interrupt the collector.
      dnCollector.interrupt();
      // Interrupt all preload threads.
      for (Thread thread : preloadThreads) {
        thread.interrupt();
      }
      logError(WARN_CACHE_PRELOAD_INTERRUPTED.get());
    } finally {
      // Kill the task in case of exception.
      timer.cancel();
    }
  }

  /**
   * The worker thread.
   */
  private class EntryCachePreloadWorker extends DirectoryThread {
    public EntryCachePreloadWorker() {
      super("Entry Cache Preload Worker");
    }
    @Override
    public void run() {
      while (!dnQueue.isEmpty() || dnCollector.isAlive()) {
        // Check if interrupted.
        if (Thread.interrupted()) {
          break;
        }
        if (interruptFlag.compareAndSet(true, false)) {
          break;
        }
        // Dequeue the next entry DN.
        try {
          DN entryDN = dnQueue.take();
          Lock readLock = null;
          try {
            // Acquire a read lock on the entry.
            readLock = LockManager.lockRead(entryDN);
            if (readLock == null) {
              // It is cheaper to put this DN back on the
              // queue then pick it up and process later.
              dnQueue.add(entryDN);
              continue;
            }
            // Even if getEntry() below fails the entry is
            // still treated as a processed entry anyways.
            processedEntries.getAndIncrement();
            // getEntry() will trigger putEntryIfAbsent() to the
            // cache if given entry is not in the cache already.
            DirectoryServer.getEntry(entryDN);
          } catch (DirectoryException ex) {
            if (debugEnabled()) {
              TRACER.debugCaught(DebugLogLevel.ERROR, ex);
            }
            Message message = ERR_CACHE_PRELOAD_ENTRY_FAILED.get(
              entryDN.toNormalizedString(),
              (ex.getCause() != null ? ex.getCause().getMessage() :
                stackTraceToSingleLineString(ex)));
            logError(message);
          } finally {
            LockManager.unlock(entryDN, readLock);
          }
        } catch (InterruptedException ex) {
          break;
        }
      }
      preloadThreads.remove(Thread.currentThread());
    }
  }

  /**
   * The Collector thread.
   */
  private class EntryCacheDNCollector extends DirectoryThread {
    public EntryCacheDNCollector() {
      super("Entry Cache Preload Collector");
    }
    @Override
    public void run() {
      Map<DN, Backend> baseDNMap =
        DirectoryServer.getPublicNamingContexts();
      Set<Backend> proccessedBackends = new HashSet<Backend>();
      // Collect all DNs from every active public backend.
      for (Backend backend : baseDNMap.values()) {
        // Check if interrupted.
        if (Thread.interrupted()) {
          return;
        }
        if (!proccessedBackends.contains(backend)) {
          proccessedBackends.add(backend);
          try {
            if (!backend.collectStoredDNs(dnQueue)) {
              // DN collection is incomplete, likely
              // due to some backend problem occured.
              // Log an error message and carry on.
              Message message =
                ERR_CACHE_PRELOAD_COLLECTOR_FAILED.get(
                backend.getBackendID());
              logError(message);
            }
          } catch (UnsupportedOperationException ex) {
            // Some backends dont have collectStoredDNs()
            // method implemented, log a warning, skip
            // such backend and continue.
            if (debugEnabled()) {
              TRACER.debugCaught(DebugLogLevel.ERROR, ex);
            }
            Message message =
              WARN_CACHE_PRELOAD_BACKEND_FAILED.get(
              backend.getBackendID());
            logError(message);
          }
        }
      }
    }
  }

  /**
   * {@inheritDoc}
   */
  public String getShutdownListenerName() {
    return preloadArbiterThreadName;
  }

  /**
   * {@inheritDoc}
   */
  public void processServerShutdown(Message reason) {
    if ((preloadArbiterThread != null) &&
         preloadArbiterThread.isAlive()) {
      // Interrupt the arbiter so it can interrupt
      // the collector and all spawned workers.
      preloadArbiterThread.interrupt();
      try {
        // This should be quick although if it
        // gets interrupted it is no big deal.
        preloadArbiterThread.join();
      } catch (InterruptedException ex) {
        if (debugEnabled()) {
          TRACER.debugCaught(DebugLogLevel.ERROR, ex);
        }
      }
    }
    DirectoryServer.deregisterShutdownListener(this);
  }
}