/*
 * 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.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.function.IntSupplier;

import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.AccumuloConfiguration.ScanExecutorConfig;
import org.apache.accumulo.core.conf.ConfigurationTypeHelper;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.file.blockfile.cache.impl.BlockCacheConfiguration;
import org.apache.accumulo.core.file.blockfile.cache.impl.BlockCacheManagerFactory;
import org.apache.accumulo.core.file.blockfile.impl.ScanCacheProvider;
import org.apache.accumulo.core.spi.cache.BlockCache;
import org.apache.accumulo.core.spi.cache.BlockCacheManager;
import org.apache.accumulo.core.spi.cache.CacheType;
import org.apache.accumulo.core.spi.common.ServiceEnvironment;
import org.apache.accumulo.core.spi.scan.ScanDirectives;
import org.apache.accumulo.core.spi.scan.ScanDispatcher;
import org.apache.accumulo.core.spi.scan.ScanDispatcher.DispatchParameters;
import org.apache.accumulo.core.spi.scan.ScanExecutor;
import org.apache.accumulo.core.spi.scan.ScanInfo;
import org.apache.accumulo.core.spi.scan.ScanPrioritizer;
import org.apache.accumulo.core.spi.scan.SimpleScanDispatcher;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.core.util.threads.Threads;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.ServiceEnvironmentImpl;
import org.apache.accumulo.tserver.FileManager.ScanFileManager;
import org.apache.accumulo.tserver.memory.LargestFirstMemoryManager;
import org.apache.accumulo.tserver.memory.TabletMemoryReport;
import org.apache.accumulo.tserver.session.ScanSession;
import org.apache.accumulo.tserver.tablet.Tablet;
import org.apache.htrace.wrappers.TraceRunnable;
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;
import com.google.common.collect.ImmutableMap;

import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;

/**
 * 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 splitThreadPool;
  private final ExecutorService defaultSplitThreadPool;
  private final ExecutorService defaultMigrationPool;
  private final ExecutorService migrationPool;
  private final ExecutorService assignmentPool;
  private final ExecutorService assignMetaDataPool;
  private final ExecutorService summaryRetrievalPool;
  private final ExecutorService summaryParitionPool;
  private final ExecutorService summaryRemotePool;

  private final Map<String,ExecutorService> scanExecutors;
  private final Map<String,ScanExecutor> scanExecutorChoices;

  private final ConcurrentHashMap<KeyExtent,RunnableStartedAt> activeAssignments;

  private final FileManager fileManager;

  private final LargestFirstMemoryManager memoryManager;

  private final MemoryManagementFramework memMgmt;

  private final BlockCacheManager cacheManager;
  private final BlockCache _dCache;
  private final BlockCache _iCache;
  private final BlockCache _sCache;
  private final ServerContext context;

  private Cache<String,Long> fileLenCache;

  /**
   * This method creates a task that changes the number of core and maximum threads on the thread
   * pool executor
   *
   * @param maxThreads
   *          max threads
   * @param name
   *          name of thread pool
   * @param tp
   *          executor
   */
  private void modifyThreadPoolSizesAtRuntime(IntSupplier maxThreads, String name,
      final ThreadPoolExecutor tp) {
    ThreadPools.createGeneralScheduledExecutorService(context.getConfiguration())
        .scheduleWithFixedDelay(() -> {
          try {
            int max = maxThreads.getAsInt();
            int currentMax = tp.getMaximumPoolSize();
            if (currentMax != max) {
              log.info("Changing max threads for {} from {} to {}", 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 (Exception t) {
            log.error("Failed to change thread pool size", t);
          }
        }, 1000, 10_000, TimeUnit.MILLISECONDS);
  }

  private ExecutorService createPriorityExecutor(ScanExecutorConfig sec,
      Map<String,Queue<?>> scanExecQueues) {

    BlockingQueue<Runnable> queue;

    if (sec.prioritizerClass.orElse("").isEmpty()) {
      queue = new LinkedBlockingQueue<>();
    } else {
      ScanPrioritizer factory = null;
      try {
        factory = ConfigurationTypeHelper.getClassInstance(null, sec.prioritizerClass.get(),
            ScanPrioritizer.class);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }

      if (factory == null) {
        queue = new LinkedBlockingQueue<>();
      } else {
        Comparator<ScanInfo> comparator =
            factory.createComparator(new ScanPrioritizer.CreateParameters() {

              @Override
              public Map<String,String> getOptions() {
                return sec.prioritizerOpts;
              }

              @Override
              public ServiceEnvironment getServiceEnv() {
                return new ServiceEnvironmentImpl(context);
              }
            });

        // function to extract scan scan session from runnable
        Function<Runnable,ScanInfo> extractor =
            r -> ((ScanSession.ScanMeasurer) ((TraceRunnable) r).getRunnable()).getScanInfo();

        queue = new PriorityBlockingQueue<>(sec.maxThreads,
            Comparator.comparing(extractor, comparator));
      }
    }

    scanExecQueues.put(sec.name, queue);

    ExecutorService es =
        ThreadPools.createThreadPool(sec.getCurrentMaxThreads(), sec.getCurrentMaxThreads(), 0L,
            TimeUnit.MILLISECONDS, "scan-" + sec.name, queue, sec.priority, true);
    modifyThreadPoolSizesAtRuntime(sec::getCurrentMaxThreads, "scan-" + sec.name,
        (ThreadPoolExecutor) es);
    return es;

  }

  protected Map<String,ExecutorService> createScanExecutors(
      Collection<ScanExecutorConfig> scanExecCfg, Map<String,Queue<?>> scanExecQueues) {
    var builder = ImmutableMap.<String,ExecutorService>builder();

    for (ScanExecutorConfig sec : scanExecCfg) {
      builder.put(sec.name, createPriorityExecutor(sec, scanExecQueues));
    }

    return builder.build();
  }

  private static class ScanExecutorImpl implements ScanExecutor {

    private static class ConfigImpl implements ScanExecutor.Config {

      final ScanExecutorConfig cfg;

      public ConfigImpl(ScanExecutorConfig sec) {
        this.cfg = sec;
      }

      @Override
      public String getName() {
        return cfg.name;
      }

      @Override
      public int getMaxThreads() {
        return cfg.maxThreads;
      }

      @Override
      public Optional<String> getPrioritizerClass() {
        return cfg.prioritizerClass;
      }

      @Override
      public Map<String,String> getPrioritizerOptions() {
        return cfg.prioritizerOpts;
      }

    }

    private final ConfigImpl config;
    private final Queue<?> queue;

    ScanExecutorImpl(ScanExecutorConfig sec, Queue<?> q) {
      this.config = new ConfigImpl(sec);
      this.queue = q;
    }

    @Override
    public int getQueued() {
      return queue.size();
    }

    @Override
    public Config getConfig() {
      return config;
    }

  }

  private Map<String,ScanExecutor> createScanExecutorChoices(
      Collection<ScanExecutorConfig> scanExecCfg, Map<String,Queue<?>> scanExecQueues) {
    var builder = ImmutableMap.<String,ScanExecutor>builder();

    for (ScanExecutorConfig sec : scanExecCfg) {
      builder.put(sec.name, new ScanExecutorImpl(sec, scanExecQueues.get(sec.name)));
    }

    return builder.build();
  }

  @SuppressFBWarnings(value = "DM_GC",
      justification = "GC is run to get a good estimate of memory availability")
  public TabletServerResourceManager(ServerContext context) {
    this.context = context;
    final AccumuloConfiguration acuConf = context.getConfiguration();

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

    long totalQueueSize = acuConf.getAsBytes(Property.TSERV_TOTAL_MUTATION_QUEUE_MAX);

    try {
      cacheManager = BlockCacheManagerFactory.getInstance(acuConf);
    } catch (Exception e) {
      throw new RuntimeException("Error creating BlockCacheManager", e);
    }

    cacheManager.start(new BlockCacheConfiguration(acuConf));

    _iCache = cacheManager.getBlockCache(CacheType.INDEX);
    _dCache = cacheManager.getBlockCache(CacheType.DATA);
    _sCache = cacheManager.getBlockCache(CacheType.SUMMARY);

    long dCacheSize = _dCache.getMaxHeapSize();
    long iCacheSize = _iCache.getMaxHeapSize();
    long sCacheSize = _sCache.getMaxHeapSize();

    Runtime runtime = Runtime.getRuntime();
    if (usingNativeMap) {
      // Still check block cache sizes when using native maps.
      if (dCacheSize + iCacheSize + sCacheSize + 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 + sCacheSize, totalQueueSize, runtime.maxMemory()));
      }
    } else if (maxMemory + dCacheSize + iCacheSize + sCacheSize + 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 + sCacheSize, 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 =
        ThreadPools.createExecutorService(acuConf, Property.TSERV_MINC_MAXCONCURRENT);
    modifyThreadPoolSizesAtRuntime(
        () -> context.getConfiguration().getCount(Property.TSERV_MINC_MAXCONCURRENT),
        "minor compactor", (ThreadPoolExecutor) minorCompactionThreadPool);

    splitThreadPool = ThreadPools.createThreadPool(0, 1, 1, TimeUnit.SECONDS, "splitter", true);

    defaultSplitThreadPool =
        ThreadPools.createThreadPool(0, 1, 60, TimeUnit.SECONDS, "md splitter", true);

    defaultMigrationPool =
        ThreadPools.createThreadPool(0, 1, 60, TimeUnit.SECONDS, "metadata tablet migration", true);

    migrationPool =
        ThreadPools.createExecutorService(acuConf, Property.TSERV_MIGRATE_MAXCONCURRENT);
    modifyThreadPoolSizesAtRuntime(
        () -> context.getConfiguration().getCount(Property.TSERV_MIGRATE_MAXCONCURRENT),
        "tablet migration", (ThreadPoolExecutor) migrationPool);

    // 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 =
        ThreadPools.createExecutorService(acuConf, Property.TSERV_ASSIGNMENT_MAXCONCURRENT);
    modifyThreadPoolSizesAtRuntime(
        () -> context.getConfiguration().getCount(Property.TSERV_ASSIGNMENT_MAXCONCURRENT),
        "tablet assignment", (ThreadPoolExecutor) assignmentPool);

    assignMetaDataPool = ThreadPools.createThreadPool(0, 1, 60, TimeUnit.SECONDS,
        "metadata tablet assignment", true);

    activeAssignments = new ConcurrentHashMap<>();

    summaryRetrievalPool =
        ThreadPools.createExecutorService(acuConf, Property.TSERV_SUMMARY_RETRIEVAL_THREADS);
    modifyThreadPoolSizesAtRuntime(
        () -> context.getConfiguration().getCount(Property.TSERV_SUMMARY_RETRIEVAL_THREADS),
        "summary file retriever", (ThreadPoolExecutor) summaryRetrievalPool);

    summaryRemotePool =
        ThreadPools.createExecutorService(acuConf, Property.TSERV_SUMMARY_REMOTE_THREADS);
    modifyThreadPoolSizesAtRuntime(
        () -> context.getConfiguration().getCount(Property.TSERV_SUMMARY_REMOTE_THREADS),
        "summary remote", (ThreadPoolExecutor) summaryRemotePool);

    summaryParitionPool =
        ThreadPools.createExecutorService(acuConf, Property.TSERV_SUMMARY_PARTITION_THREADS);
    modifyThreadPoolSizesAtRuntime(
        () -> context.getConfiguration().getCount(Property.TSERV_SUMMARY_PARTITION_THREADS),
        "summary partition", (ThreadPoolExecutor) summaryParitionPool);

    Collection<ScanExecutorConfig> scanExecCfg = acuConf.getScanExecutors();
    Map<String,Queue<?>> scanExecQueues = new HashMap<>();
    scanExecutors = createScanExecutors(scanExecCfg, scanExecQueues);
    scanExecutorChoices = createScanExecutorChoices(scanExecCfg, scanExecQueues);

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

    fileLenCache =
        CacheBuilder.newBuilder().maximumSize(Math.min(maxOpenFiles * 1000L, 100_000)).build();

    fileManager = new FileManager(context, maxOpenFiles, fileLenCache);

    memoryManager = new LargestFirstMemoryManager();
    memoryManager.init(context);
    memMgmt = new MemoryManagementFramework();
    memMgmt.startThreads();

    // 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.
    ThreadPools.createGeneralScheduledExecutorService(context.getConfiguration())
        .schedule(new AssignmentWatcher(acuConf, activeAssignments), 5000, TimeUnit.MILLISECONDS);
  }

  /**
   * 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;

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

    @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 {} has been running for at least {}ms", extent, duration,
                runnable.getTask().getException());
          } else if (log.isTraceEnabled()) {
            log.trace("Assignment for {} only running for {}ms", extent, duration);
          }
        }
      } 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 {}ms", delay);
        }
        ThreadPools.createGeneralScheduledExecutorService(conf).schedule(this, delay,
            TimeUnit.MILLISECONDS);
      }
    }
  }

  private class MemoryManagementFramework {
    private final Map<KeyExtent,TabletMemoryReport> tabletReports;
    private final LinkedBlockingQueue<TabletMemoryReport> 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<>());
      memUsageReports = new LinkedBlockingQueue<>();
      maxMem = context.getConfiguration().getAsBytes(Property.TSERV_MAXMEM);
      memoryGuardThread = Threads.createThread("Accumulo Memory Guard",
          OptionalInt.of(Thread.NORM_PRIORITY + 1), this::processTabletMemStats);
      minorCompactionInitiatorThread =
          Threads.createThread("Accumulo Minor Compaction Initiator", this::manageMemory);
    }

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

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

          TabletMemoryReport 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 (TabletMemoryReport 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) {
        List<KeyExtent> tabletsToMinorCompact = null;

        Map<KeyExtent,TabletMemoryReport> tabletReportsCopy = null;
        try {
          synchronized (tabletReports) {
            tabletReportsCopy = new HashMap<>(tabletReports);
          }
          ArrayList<TabletMemoryReport> tabletStates = new ArrayList<>(tabletReportsCopy.values());
          tabletsToMinorCompact = memoryManager.tabletsToMinorCompact(tabletStates);

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

        try {
          if (tabletsToMinorCompact != null && !tabletsToMinorCompact.isEmpty()) {
            for (KeyExtent keyExtent : tabletsToMinorCompact) {
              TabletMemoryReport tabletReport = tabletReportsCopy.get(keyExtent);

              if (tabletReport == null) {
                log.warn("Memory manager asked to compact nonexistent tablet"
                    + " {}; manager implementation might be misbehaving", keyExtent);
                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) {
                    TabletMemoryReport latestReport = tabletReports.remove(keyExtent);
                    if (latestReport != null) {
                      if (latestReport.getTablet() == tablet) {
                        log.debug("Cleaned up report for closed tablet {}", keyExtent);
                      } else {
                        // different tablet instance => put it back
                        tabletReports.put(keyExtent, latestReport);
                      }
                    }
                  }
                  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 (Exception t) {
          log.error("Minor compactions for memory management failed", t);
        }

        sleepUninterruptibly(250, TimeUnit.MILLISECONDS);
      }
    }

    public void updateMemoryUsageStats(Tablet tablet, long size, long lastCommitTime,
        long mincSize) {
      memUsageReports.add(new TabletMemoryReport(tablet, lastCommitTime, size, 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()
          + context.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 synchronized TabletResourceManager createTabletResourceManager(KeyExtent extent,
      AccumuloConfiguration conf) {
    return new TabletResourceManager(extent, conf);
  }

  public class TabletResourceManager {

    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(ScanDirectives scanDirectives) {
      if (closed) {
        throw new IllegalStateException("closed");
      }

      return fileManager.newScanFileManager(extent,
          new ScanCacheProvider(tableConf, scanDirectives, _iCache, _dCache));
    }

    // 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

    // 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(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);

          closed = true;
        }
      }
    }

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

  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);
    }
  }

  @SuppressWarnings("deprecation")
  private static abstract class DispatchParamsImpl implements DispatchParameters,
      org.apache.accumulo.core.spi.scan.ScanDispatcher.DispatchParmaters {

  }

  public void executeReadAhead(KeyExtent tablet, ScanDispatcher dispatcher, ScanSession scanInfo,
      Runnable task) {

    task = ScanSession.wrap(scanInfo, task);

    if (tablet.isRootTablet()) {
      // TODO make meta dispatch??
      scanInfo.scanParams.setScanDirectives(ScanDirectives.builder().build());
      task.run();
    } else if (tablet.isMeta()) {
      // TODO make meta dispatch??
      scanInfo.scanParams.setScanDirectives(ScanDirectives.builder().build());
      scanExecutors.get("meta").execute(task);
    } else {
      DispatchParameters params = new DispatchParamsImpl() {
        @Override
        public ScanInfo getScanInfo() {
          return scanInfo;
        }

        @Override
        public Map<String,ScanExecutor> getScanExecutors() {
          return scanExecutorChoices;
        }

        @Override
        public ServiceEnvironment getServiceEnv() {
          return new ServiceEnvironmentImpl(context);
        }
      };

      ScanDirectives prefs = dispatcher.dispatch(params);
      scanInfo.scanParams.setScanDirectives(prefs);

      ExecutorService executor = scanExecutors.get(prefs.getExecutorName());
      if (executor == null) {
        log.warn(
            "For table id {}, {} dispatched to non-existant executor {} Using default executor.",
            tablet.tableId(), dispatcher.getClass().getName(), prefs.getExecutorName());
        executor = scanExecutors.get(SimpleScanDispatcher.DEFAULT_SCAN_EXECUTOR_NAME);
      } else if ("meta".equals(prefs.getExecutorName())) {
        log.warn("For table id {}, {} dispatched to meta executor. Using default executor.",
            tablet.tableId(), dispatcher.getClass().getName());
        executor = scanExecutors.get(SimpleScanDispatcher.DEFAULT_SCAN_EXECUTOR_NAME);
      }
      executor.execute(task);
    }
  }

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

  public void addMetaDataAssignment(KeyExtent extent, Logger log,
      AssignmentHandler assignmentHandler) {
    assignMetaDataPool
        .execute(new ActiveAssignmentRunnable(activeAssignments, extent, 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 BlockCache getIndexCache() {
    return _iCache;
  }

  public BlockCache getDataCache() {
    return _dCache;
  }

  public BlockCache getSummaryCache() {
    return _sCache;
  }

  public Cache<String,Long> getFileLenCache() {
    return fileLenCache;
  }

  public ExecutorService getSummaryRetrievalExecutor() {
    return summaryRetrievalPool;
  }

  public ExecutorService getSummaryPartitionExecutor() {
    return summaryParitionPool;
  }

  public ExecutorService getSummaryRemoteExecutor() {
    return summaryRemotePool;
  }
}