/*
 * 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.accumulo.tserver;

import static java.util.Objects.requireNonNull;
import static org.apache.accumulo.fate.util.UtilWaitThread.sleepUninterruptibly;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.impl.KeyExtent;
import org.apache.accumulo.core.file.blockfile.cache.LruBlockCache;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.util.Daemon;
import org.apache.accumulo.core.util.NamingThreadFactory;
import org.apache.accumulo.fate.util.LoggingRunnable;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.fs.FileRef;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.tabletserver.LargestFirstMemoryManager;
import org.apache.accumulo.server.tabletserver.MemoryManagementActions;
import org.apache.accumulo.server.tabletserver.MemoryManager;
import org.apache.accumulo.server.tabletserver.TabletState;
import org.apache.accumulo.server.util.time.SimpleTimer;
import org.apache.accumulo.tserver.FileManager.ScanFileManager;
import org.apache.accumulo.tserver.TabletServer.AssignmentHandler;
import org.apache.accumulo.tserver.compaction.CompactionStrategy;
import org.apache.accumulo.tserver.compaction.DefaultCompactionStrategy;
import org.apache.accumulo.tserver.compaction.MajorCompactionReason;
import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
import org.apache.accumulo.tserver.tablet.Tablet;
import org.apache.htrace.wrappers.TraceExecutorService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;

/**
 * ResourceManager is responsible for managing the resources of all tablets within a tablet server.
 */
public class TabletServerResourceManager {

  private static final Logger log = LoggerFactory.getLogger(TabletServerResourceManager.class);

  private final ExecutorService minorCompactionThreadPool;
  private final ExecutorService majorCompactionThreadPool;
  private final ExecutorService rootMajorCompactionThreadPool;
  private final ExecutorService defaultMajorCompactionThreadPool;
  private final ExecutorService splitThreadPool;
  private final ExecutorService defaultSplitThreadPool;
  private final ExecutorService defaultMigrationPool;
  private final ExecutorService migrationPool;
  private final ExecutorService assignmentPool;
  private final ExecutorService assignMetaDataPool;
  private final ExecutorService readAheadThreadPool;
  private final ExecutorService defaultReadAheadThreadPool;
  private final Map<String,ExecutorService> threadPools = new TreeMap<>();

  private final ConcurrentHashMap<KeyExtent,RunnableStartedAt> activeAssignments;

  private final VolumeManager fs;

  private final FileManager fileManager;

  private final MemoryManager memoryManager;

  private final MemoryManagementFramework memMgmt;

  private final LruBlockCache _dCache;
  private final LruBlockCache _iCache;
  private final TabletServer tserver;
  private final ServerConfigurationFactory conf;

  private ExecutorService addEs(String name, ExecutorService tp) {
    if (threadPools.containsKey(name)) {
      throw new IllegalArgumentException(
          "Cannot create two executor services with same name " + name);
    }
    tp = new TraceExecutorService(tp);
    threadPools.put(name, tp);
    return tp;
  }

  private ExecutorService addEs(final Property maxThreads, final String name,
      final ThreadPoolExecutor tp) {
    ExecutorService result = addEs(name, tp);
    SimpleTimer.getInstance(tserver.getConfiguration()).schedule(new Runnable() {
      @Override
      public void run() {
        try {
          int max = tserver.getConfiguration().getCount(maxThreads);
          int currentMax = tp.getMaximumPoolSize();
          if (currentMax != max) {
            log.info("Changing {} for {} from {} to {}", maxThreads.getKey(), name, currentMax,
                max);
            if (max > currentMax) {
              // increasing, increase the max first, or the core will fail to be increased
              tp.setMaximumPoolSize(max);
              tp.setCorePoolSize(max);
            } else {
              // decreasing, lower the core size first, or the max will fail to be lowered
              tp.setCorePoolSize(max);
              tp.setMaximumPoolSize(max);
            }
          }
        } catch (Throwable t) {
          log.error("Failed to change thread pool size", t);
        }
      }

    }, 1000, 10_000);
    return result;
  }

  private ExecutorService createEs(int max, String name) {
    return addEs(name, Executors.newFixedThreadPool(max, new NamingThreadFactory(name)));
  }

  private ExecutorService createEs(Property max, String name) {
    return createEs(max, name, new LinkedBlockingQueue<Runnable>());
  }

  private ExecutorService createEs(Property max, String name, BlockingQueue<Runnable> queue) {
    int maxThreads = conf.getConfiguration().getCount(max);
    ThreadPoolExecutor tp = new ThreadPoolExecutor(maxThreads, maxThreads, 0L,
        TimeUnit.MILLISECONDS, queue, new NamingThreadFactory(name));
    return addEs(max, name, tp);
  }

  private ExecutorService createEs(int min, int max, int timeout, String name) {
    return addEs(name, new ThreadPoolExecutor(min, max, timeout, TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(), new NamingThreadFactory(name)));
  }

  public TabletServerResourceManager(TabletServer tserver, VolumeManager fs) {
    this.tserver = tserver;
    this.conf = tserver.getServerConfigurationFactory();
    this.fs = fs;
    final AccumuloConfiguration acuConf = conf.getConfiguration();

    long maxMemory = acuConf.getMemoryInBytes(Property.TSERV_MAXMEM);
    boolean usingNativeMap =
        acuConf.getBoolean(Property.TSERV_NATIVEMAP_ENABLED) && NativeMap.isLoaded();

    long blockSize = acuConf.getMemoryInBytes(Property.TSERV_DEFAULT_BLOCKSIZE);
    long dCacheSize = acuConf.getMemoryInBytes(Property.TSERV_DATACACHE_SIZE);
    long iCacheSize = acuConf.getMemoryInBytes(Property.TSERV_INDEXCACHE_SIZE);
    long totalQueueSize = acuConf.getMemoryInBytes(Property.TSERV_TOTAL_MUTATION_QUEUE_MAX);

    _iCache = new LruBlockCache(iCacheSize, blockSize);
    _dCache = new LruBlockCache(dCacheSize, blockSize);

    Runtime runtime = Runtime.getRuntime();
    if (usingNativeMap) {
      // Still check block cache sizes when using native maps.
      if (dCacheSize + iCacheSize + totalQueueSize > runtime.maxMemory()) {
        throw new IllegalArgumentException(String.format(
            "Block cache sizes %,d"
                + " and mutation queue size %,d is too large for this JVM configuration %,d",
            dCacheSize + iCacheSize, totalQueueSize, runtime.maxMemory()));
      }
    } else if (maxMemory + dCacheSize + iCacheSize + totalQueueSize > runtime.maxMemory()) {
      throw new IllegalArgumentException(String.format(
          "Maximum tablet server"
              + " map memory %,d block cache sizes %,d and mutation queue size %,d is"
              + " too large for this JVM configuration %,d",
          maxMemory, dCacheSize + iCacheSize, totalQueueSize, runtime.maxMemory()));
    }
    runtime.gc();

    // totalMemory - freeMemory = memory in use
    // maxMemory - memory in use = max available memory
    if (!usingNativeMap
        && maxMemory > runtime.maxMemory() - (runtime.totalMemory() - runtime.freeMemory())) {
      log.warn("In-memory map may not fit into local memory space.");
    }

    minorCompactionThreadPool = createEs(Property.TSERV_MINC_MAXCONCURRENT, "minor compactor");

    // make this thread pool have a priority queue... and execute tablets with the most
    // files first!
    majorCompactionThreadPool = createEs(Property.TSERV_MAJC_MAXCONCURRENT, "major compactor",
        new CompactionQueue().asBlockingQueueOfRunnable());
    rootMajorCompactionThreadPool = createEs(0, 1, 300, "md root major compactor");
    defaultMajorCompactionThreadPool = createEs(0, 1, 300, "md major compactor");

    splitThreadPool = createEs(1, "splitter");
    defaultSplitThreadPool = createEs(0, 1, 60, "md splitter");

    defaultMigrationPool = createEs(0, 1, 60, "metadata tablet migration");
    migrationPool = createEs(Property.TSERV_MIGRATE_MAXCONCURRENT, "tablet migration");

    // not sure if concurrent assignments can run safely... even if they could there is probably no
    // benefit at startup because
    // individual tablet servers are already running assignments concurrently... having each
    // individual tablet server run
    // concurrent assignments would put more load on the metadata table at startup
    assignmentPool = createEs(Property.TSERV_ASSIGNMENT_MAXCONCURRENT, "tablet assignment");

    assignMetaDataPool = createEs(0, 1, 60, "metadata tablet assignment");

    activeAssignments = new ConcurrentHashMap<>();

    readAheadThreadPool = createEs(Property.TSERV_READ_AHEAD_MAXCONCURRENT, "tablet read ahead");
    defaultReadAheadThreadPool =
        createEs(Property.TSERV_METADATA_READ_AHEAD_MAXCONCURRENT, "metadata tablets read ahead");

    int maxOpenFiles = acuConf.getCount(Property.TSERV_SCAN_MAX_OPENFILES);

    Cache<String,Long> fileLenCache =
        CacheBuilder.newBuilder().maximumSize(Math.min(maxOpenFiles * 1000L, 100_000)).build();

    fileManager = new FileManager(tserver, fs, maxOpenFiles, fileLenCache, _dCache, _iCache);

    memoryManager = Property.createInstanceFromPropertyName(acuConf, Property.TSERV_MEM_MGMT,
        MemoryManager.class, new LargestFirstMemoryManager());
    memoryManager.init(tserver.getServerConfigurationFactory());
    memMgmt = new MemoryManagementFramework();
    memMgmt.startThreads();

    SimpleTimer timer = SimpleTimer.getInstance(tserver.getConfiguration());

    // We can use the same map for both metadata and normal assignments since the keyspace (extent)
    // is guaranteed to be unique. Schedule the task once, the task will reschedule itself.
    timer.schedule(new AssignmentWatcher(acuConf, activeAssignments, timer), 5000);
  }

  /**
   * Accepts some map which is tracking active assignment task(s) (running) and monitors them to
   * ensure that the time the assignment(s) have been running don't exceed a threshold. If the time
   * is exceeded a warning is printed and a stack trace is logged for the running assignment.
   */
  protected static class AssignmentWatcher implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(AssignmentWatcher.class);

    private final Map<KeyExtent,RunnableStartedAt> activeAssignments;
    private final AccumuloConfiguration conf;
    private final SimpleTimer timer;

    public AssignmentWatcher(AccumuloConfiguration conf,
        Map<KeyExtent,RunnableStartedAt> activeAssignments, SimpleTimer timer) {
      this.conf = conf;
      this.activeAssignments = activeAssignments;
      this.timer = timer;
    }

    @Override
    public void run() {
      final long millisBeforeWarning =
          conf.getTimeInMillis(Property.TSERV_ASSIGNMENT_DURATION_WARNING);
      try {
        long now = System.currentTimeMillis();
        KeyExtent extent;
        RunnableStartedAt runnable;
        for (Entry<KeyExtent,RunnableStartedAt> entry : activeAssignments.entrySet()) {
          extent = entry.getKey();
          runnable = entry.getValue();
          final long duration = now - runnable.getStartTime();

          // Print a warning if an assignment has been running for over the configured time length
          if (duration > millisBeforeWarning) {
            log.warn(
                "Assignment for " + extent + " has been running for at least " + duration + "ms",
                runnable.getTask().getException());
          } else if (log.isTraceEnabled()) {
            log.trace("Assignment for " + extent + " only running for " + duration + "ms");
          }
        }
      } catch (Exception e) {
        log.warn("Caught exception checking active assignments", e);
      } finally {
        // Don't run more often than every 5s
        long delay = Math.max((long) (millisBeforeWarning * 0.5), 5000l);
        if (log.isTraceEnabled()) {
          log.trace("Rescheduling assignment watcher to run in " + delay + "ms");
        }
        timer.schedule(this, delay);
      }
    }
  }

  private static class TabletStateImpl implements TabletState, Cloneable {

    private final long lct;
    private final Tablet tablet;
    private final long mts;
    private final long mcmts;

    public TabletStateImpl(Tablet t, long mts, long lct, long mcmts) {
      this.tablet = t;
      this.mts = mts;
      this.lct = lct;
      this.mcmts = mcmts;
    }

    @Override
    public KeyExtent getExtent() {
      return tablet.getExtent();
    }

    Tablet getTablet() {
      return tablet;
    }

    @Override
    public long getLastCommitTime() {
      return lct;
    }

    @Override
    public long getMemTableSize() {
      return mts;
    }

    @Override
    public long getMinorCompactingMemTableSize() {
      return mcmts;
    }

    @Override
    public TabletStateImpl clone() throws CloneNotSupportedException {
      return (TabletStateImpl) super.clone();
    }
  }

  private class MemoryManagementFramework {
    private final Map<KeyExtent,TabletStateImpl> tabletReports;
    private final LinkedBlockingQueue<TabletStateImpl> memUsageReports;
    private long lastMemCheckTime = System.currentTimeMillis();
    private long maxMem;
    private long lastMemTotal = 0;
    private final Thread memoryGuardThread;
    private final Thread minorCompactionInitiatorThread;

    MemoryManagementFramework() {
      tabletReports = Collections.synchronizedMap(new HashMap<KeyExtent,TabletStateImpl>());
      memUsageReports = new LinkedBlockingQueue<>();
      maxMem = conf.getConfiguration().getMemoryInBytes(Property.TSERV_MAXMEM);

      Runnable r1 = new Runnable() {
        @Override
        public void run() {
          processTabletMemStats();
        }
      };

      memoryGuardThread = new Daemon(new LoggingRunnable(log, r1));
      memoryGuardThread.setPriority(Thread.NORM_PRIORITY + 1);
      memoryGuardThread.setName("Accumulo Memory Guard");

      Runnable r2 = new Runnable() {
        @Override
        public void run() {
          manageMemory();
        }
      };

      minorCompactionInitiatorThread = new Daemon(new LoggingRunnable(log, r2));
      minorCompactionInitiatorThread.setName("Accumulo Minor Compaction Initiator");
    }

    void startThreads() {
      memoryGuardThread.start();
      minorCompactionInitiatorThread.start();
    }

    private void processTabletMemStats() {
      while (true) {
        try {

          TabletStateImpl report = memUsageReports.take();

          while (report != null) {
            tabletReports.put(report.getExtent(), report);
            report = memUsageReports.poll();
          }

          long delta = System.currentTimeMillis() - lastMemCheckTime;
          if (holdCommits || delta > 50 || lastMemTotal > 0.90 * maxMem) {
            lastMemCheckTime = System.currentTimeMillis();

            long totalMemUsed = 0;

            synchronized (tabletReports) {
              for (TabletStateImpl tsi : tabletReports.values()) {
                totalMemUsed += tsi.getMemTableSize();
                totalMemUsed += tsi.getMinorCompactingMemTableSize();
              }
            }

            if (totalMemUsed > 0.95 * maxMem) {
              holdAllCommits(true);
            } else {
              holdAllCommits(false);
            }

            lastMemTotal = totalMemUsed;
          }

        } catch (InterruptedException e) {
          log.warn("Interrupted processing tablet memory statistics", e);
        }
      }
    }

    private void manageMemory() {
      while (true) {
        MemoryManagementActions mma = null;

        Map<KeyExtent,TabletStateImpl> tabletReportsCopy = null;
        try {
          synchronized (tabletReports) {
            tabletReportsCopy = new HashMap<>(tabletReports);
          }
          ArrayList<TabletState> tabletStates =
              new ArrayList<TabletState>(tabletReportsCopy.values());
          mma = memoryManager.getMemoryManagementActions(tabletStates);

        } catch (Throwable t) {
          log.error("Memory manager failed {}", t.getMessage(), t);
        }

        try {
          if (mma != null && mma.tabletsToMinorCompact != null
              && mma.tabletsToMinorCompact.size() > 0) {
            for (KeyExtent keyExtent : mma.tabletsToMinorCompact) {
              TabletStateImpl tabletReport = tabletReportsCopy.get(keyExtent);

              if (tabletReport == null) {
                log.warn("Memory manager asked to compact nonexistent tablet " + keyExtent
                    + "; manager implementation might be misbehaving");
                continue;
              }
              Tablet tablet = tabletReport.getTablet();
              if (!tablet.initiateMinorCompaction(MinorCompactionReason.SYSTEM)) {
                if (tablet.isClosed()) {
                  // attempt to remove it from the current reports if still there
                  synchronized (tabletReports) {
                    TabletStateImpl latestReport = tabletReports.remove(keyExtent);
                    if (latestReport != null) {
                      if (latestReport.getTablet() != tablet) {
                        // different tablet instance => put it back
                        tabletReports.put(keyExtent, latestReport);
                      } else {
                        log.debug("Cleaned up report for closed tablet " + keyExtent);
                      }
                    }
                  }
                  log.debug(
                      "Ignoring memory manager recommendation: not minor compacting closed tablet "
                          + keyExtent);
                } else {
                  log.info(
                      "Ignoring memory manager recommendation: not minor compacting " + keyExtent);
                }
              }
            }

            // log.debug("mma.tabletsToMinorCompact = "+mma.tabletsToMinorCompact);
          }
        } catch (Throwable t) {
          log.error("Minor compactions for memory managment failed", t);
        }

        sleepUninterruptibly(250, TimeUnit.MILLISECONDS);
      }
    }

    public void updateMemoryUsageStats(Tablet tablet, long size, long lastCommitTime,
        long mincSize) {
      memUsageReports.add(new TabletStateImpl(tablet, size, lastCommitTime, mincSize));
    }

    public void tabletClosed(KeyExtent extent) {
      tabletReports.remove(extent);
    }
  }

  private final Object commitHold = new Object();
  private volatile boolean holdCommits = false;
  private long holdStartTime;

  protected void holdAllCommits(boolean holdAllCommits) {
    synchronized (commitHold) {
      if (holdCommits != holdAllCommits) {
        holdCommits = holdAllCommits;

        if (holdCommits) {
          holdStartTime = System.currentTimeMillis();
        }

        if (!holdCommits) {
          log.debug(String.format("Commits held for %6.2f secs",
              (System.currentTimeMillis() - holdStartTime) / 1000.0));
          commitHold.notifyAll();
        }
      }
    }

  }

  void waitUntilCommitsAreEnabled() {
    if (holdCommits) {
      long timeout = System.currentTimeMillis()
          + conf.getConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT);
      synchronized (commitHold) {
        while (holdCommits) {
          try {
            if (System.currentTimeMillis() > timeout)
              throw new HoldTimeoutException("Commits are held");
            commitHold.wait(1000);
          } catch (InterruptedException e) {}
        }
      }
    }
  }

  public long holdTime() {
    if (!holdCommits)
      return 0;
    synchronized (commitHold) {
      return System.currentTimeMillis() - holdStartTime;
    }
  }

  public void close() {
    for (ExecutorService executorService : threadPools.values()) {
      executorService.shutdown();
    }

    for (Entry<String,ExecutorService> entry : threadPools.entrySet()) {
      while (true) {
        try {
          if (entry.getValue().awaitTermination(60, TimeUnit.SECONDS))
            break;
          log.info("Waiting for thread pool " + entry.getKey() + " to shutdown");
        } catch (InterruptedException e) {
          log.warn("Interrupted waiting for executor to terminate", e);
        }
      }
    }
  }

  public synchronized TabletResourceManager createTabletResourceManager(KeyExtent extent,
      AccumuloConfiguration conf) {
    TabletResourceManager trm = new TabletResourceManager(extent, conf);
    return trm;
  }

  public class TabletResourceManager {

    private final long creationTime = System.currentTimeMillis();

    private volatile boolean openFilesReserved = false;

    private volatile boolean closed = false;

    private final KeyExtent extent;

    private final AccumuloConfiguration tableConf;

    TabletResourceManager(KeyExtent extent, AccumuloConfiguration tableConf) {
      requireNonNull(extent, "extent is null");
      requireNonNull(tableConf, "tableConf is null");
      this.extent = extent;
      this.tableConf = tableConf;
    }

    @VisibleForTesting
    KeyExtent getExtent() {
      return extent;
    }

    @VisibleForTesting
    AccumuloConfiguration getTableConfiguration() {
      return tableConf;
    }

    // BEGIN methods that Tablets call to manage their set of open map files

    public void importedMapFiles() {
      lastReportedCommitTime = System.currentTimeMillis();
    }

    public synchronized ScanFileManager newScanFileManager() {
      if (closed)
        throw new IllegalStateException("closed");
      return fileManager.newScanFileManager(extent);
    }

    // END methods that Tablets call to manage their set of open map files

    // BEGIN methods that Tablets call to manage memory

    private final AtomicLong lastReportedSize = new AtomicLong();
    private final AtomicLong lastReportedMincSize = new AtomicLong();
    private volatile long lastReportedCommitTime = 0;

    public void updateMemoryUsageStats(Tablet tablet, long size, long mincSize) {

      // do not want to update stats for every little change,
      // so only do it under certain circumstances... the reason
      // for this is that reporting stats acquires a lock, do
      // not want all tablets locking on the same lock for every
      // commit
      long totalSize = size + mincSize;
      long lrs = lastReportedSize.get();
      long delta = totalSize - lrs;
      long lrms = lastReportedMincSize.get();
      boolean report = false;
      // the atomic longs are considered independently, when one is set
      // the other is not set intentionally because this method is not
      // synchronized... therefore there are not transactional semantics
      // for reading and writing two variables
      if ((lrms > 0 && mincSize == 0 || lrms == 0 && mincSize > 0)
          && lastReportedMincSize.compareAndSet(lrms, mincSize)) {
        report = true;
      }

      long currentTime = System.currentTimeMillis();
      if ((delta > 32000 || delta < 0 || (currentTime - lastReportedCommitTime > 1000))
          && lastReportedSize.compareAndSet(lrs, totalSize)) {
        if (delta > 0)
          lastReportedCommitTime = currentTime;
        report = true;
      }

      if (report)
        memMgmt.updateMemoryUsageStats(tablet, size, lastReportedCommitTime, mincSize);
    }

    // END methods that Tablets call to manage memory

    // BEGIN methods that Tablets call to make decisions about major compaction
    // when too many files are open, we may want tablets to compact down
    // to one map file
    public boolean needsMajorCompaction(SortedMap<FileRef,DataFileValue> tabletFiles,
        MajorCompactionReason reason) {
      if (closed)
        return false;// throw new IOException("closed");

      // int threshold;

      if (reason == MajorCompactionReason.USER)
        return true;

      if (reason == MajorCompactionReason.IDLE) {
        // threshold = 1;
        long idleTime;
        if (lastReportedCommitTime == 0) {
          // no commits, so compute how long the tablet has been assigned to the
          // tablet server
          idleTime = System.currentTimeMillis() - creationTime;
        } else {
          idleTime = System.currentTimeMillis() - lastReportedCommitTime;
        }

        if (idleTime < tableConf.getTimeInMillis(Property.TABLE_MAJC_COMPACTALL_IDLETIME)) {
          return false;
        }
      }
      CompactionStrategy strategy = Property.createTableInstanceFromPropertyName(tableConf,
          Property.TABLE_COMPACTION_STRATEGY, CompactionStrategy.class,
          new DefaultCompactionStrategy());
      strategy.init(Property.getCompactionStrategyOptions(tableConf));
      MajorCompactionRequest request = new MajorCompactionRequest(extent, reason,
          TabletServerResourceManager.this.fs, tableConf);
      request.setFiles(tabletFiles);
      try {
        return strategy.shouldCompact(request);
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }

    // END methods that Tablets call to make decisions about major compaction

    // tablets call this method to run minor compactions,
    // this allows us to control how many minor compactions
    // run concurrently in a tablet server
    public void executeMinorCompaction(final Runnable r) {
      minorCompactionThreadPool.execute(new LoggingRunnable(log, r));
    }

    public void close() throws IOException {
      // always obtain locks in same order to avoid deadlock
      synchronized (TabletServerResourceManager.this) {
        synchronized (this) {
          if (closed)
            throw new IOException("closed");
          if (openFilesReserved)
            throw new IOException("tired to close files while open files reserved");

          memMgmt.tabletClosed(extent);
          memoryManager.tabletClosed(extent);

          closed = true;
        }
      }
    }

    public TabletServerResourceManager getTabletServerResourceManager() {
      return TabletServerResourceManager.this;
    }

    public void executeMajorCompaction(KeyExtent tablet, Runnable compactionTask) {
      TabletServerResourceManager.this.executeMajorCompaction(tablet, compactionTask);
    }

  }

  public void executeSplit(KeyExtent tablet, Runnable splitTask) {
    if (tablet.isMeta()) {
      if (tablet.isRootTablet()) {
        log.warn("Saw request to split root tablet, ignoring");
        return;
      }
      defaultSplitThreadPool.execute(splitTask);
    } else {
      splitThreadPool.execute(splitTask);
    }
  }

  public void executeMajorCompaction(KeyExtent tablet, Runnable compactionTask) {
    if (tablet.isRootTablet()) {
      rootMajorCompactionThreadPool.execute(compactionTask);
    } else if (tablet.isMeta()) {
      defaultMajorCompactionThreadPool.execute(compactionTask);
    } else {
      majorCompactionThreadPool.execute(compactionTask);
    }
  }

  public void executeReadAhead(KeyExtent tablet, Runnable task) {
    if (tablet.isRootTablet()) {
      task.run();
    } else if (tablet.isMeta()) {
      defaultReadAheadThreadPool.execute(task);
    } else {
      readAheadThreadPool.execute(task);
    }
  }

  public void addAssignment(KeyExtent extent, Logger log, AssignmentHandler assignmentHandler) {
    assignmentPool.execute(new ActiveAssignmentRunnable(activeAssignments, extent,
        new LoggingRunnable(log, assignmentHandler)));
  }

  public void addMetaDataAssignment(KeyExtent extent, Logger log,
      AssignmentHandler assignmentHandler) {
    assignMetaDataPool.execute(new ActiveAssignmentRunnable(activeAssignments, extent,
        new LoggingRunnable(log, assignmentHandler)));
  }

  public void addMigration(KeyExtent tablet, Runnable migrationHandler) {
    if (tablet.isRootTablet()) {
      migrationHandler.run();
    } else if (tablet.isMeta()) {
      defaultMigrationPool.execute(migrationHandler);
    } else {
      migrationPool.execute(migrationHandler);
    }
  }

  public LruBlockCache getIndexCache() {
    return _iCache;
  }

  public LruBlockCache getDataCache() {
    return _dCache;
  }

}