alluxio-master-2.x/core/server/worker/src/main/java/alluxio/worker/block/management/ManagementTaskCoordinator.java
/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.worker.block.management;

import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.util.ThreadFactoryUtils;
import alluxio.worker.block.BlockMetadataEvictorView;
import alluxio.worker.block.BlockMetadataManager;
import alluxio.worker.block.BlockStoreLocation;
import alluxio.worker.block.LocalBlockStore;
import alluxio.worker.block.management.tier.TierManagementTaskProvider;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Supplier;

/**
 * Coordinator for instantiating and running various block management tasks.
 */
public class ManagementTaskCoordinator implements Closeable {
  private static final Logger LOG = LoggerFactory.getLogger(ManagementTaskCoordinator.class);
  /** Duration to sleep when a) load detected on worker. b) no work to do. */
  private static final long LOAD_DETECTION_COOL_DOWN_TIME =
      Configuration.getMs(PropertyKey.WORKER_MANAGEMENT_LOAD_DETECTION_COOL_DOWN_TIME);
  /** The back-off strategy. */
  private static final BackoffStrategy BACKOFF_STRATEGY = Configuration
      .getEnum(PropertyKey.WORKER_MANAGEMENT_BACKOFF_STRATEGY, BackoffStrategy.class);

  /** Runner thread for launching management tasks. */
  private final Thread mRunnerThread;
  /** Executor that will run the management tasks. */
  private final ExecutorService mTaskExecutor = Executors.newFixedThreadPool(
      Configuration.getInt(PropertyKey.WORKER_MANAGEMENT_TASK_THREAD_COUNT),
        ThreadFactoryUtils.build("block-management-task-%d", true));

  private final LocalBlockStore mBlockStore;
  private final BlockMetadataManager mMetadataManager;
  private final StoreLoadTracker mLoadTracker;

  /** This coordinator requires calculating eviction view per each task. */
  private final Supplier<BlockMetadataEvictorView> mEvictionViewSupplier;

  /** List of management task providers. */
  private List<ManagementTaskProvider> mTaskProviders;

  /**
   * Creates management coordinator.
   *
   * @param blockStore block store
   * @param metadataManager meta manager
   * @param loadTracker load tracker
   * @param evictionViewSupplier eviction view supplier
   */
  public ManagementTaskCoordinator(LocalBlockStore blockStore, BlockMetadataManager metadataManager,
      StoreLoadTracker loadTracker, Supplier<BlockMetadataEvictorView> evictionViewSupplier) {
    mBlockStore = blockStore;
    mMetadataManager = metadataManager;
    mLoadTracker = loadTracker;
    mEvictionViewSupplier = evictionViewSupplier;
    initializeTaskProviders();
    // Initialize runner thread.
    mRunnerThread = new Thread(this::runManagement, "block-management-runner");
    mRunnerThread.setDaemon(true);
  }

  /**
   * Starts the coordinator.
   */
  public void start() {
    // Start runner thread.
    mRunnerThread.start();
  }

  @Override
  public void close() throws IOException {
    try {
      // Shutdown task executor.
      mTaskExecutor.shutdownNow();
      // Interrupt and wait for runner thread.
      mRunnerThread.interrupt();
      mRunnerThread.join();
    } catch (Exception e) {
      throw new IOException("Failed to close management task coordinator", e);
    }
  }

  /**
   * Register known task providers by priority order.
   *
   * TODO(ggezer): Re-implement async-cache as {@link BlockManagementTask}.
   */
  private void initializeTaskProviders() {
    mTaskProviders = new ArrayList<>(1);
    if (Configuration.isSet(PropertyKey.WORKER_EVICTOR_CLASS)) {
      LOG.warn("Tier management tasks will be disabled under eviction emulation mode.");
    } else {
      // TODO(ggezer): Improve on views per task type.
      mTaskProviders.add(new TierManagementTaskProvider(mBlockStore, mMetadataManager,
          mEvictionViewSupplier, mLoadTracker, mTaskExecutor));
    }
  }

  /**
   * @return the next management task to run, {@code null} if none pending
   */
  private BlockManagementTask getNextTask() {
    /*
     * Order of providers in the registered list imposes an implicit priority of tasks.
     * As long as a provider gives a task, providers next to it won't be consulted.
     */
    for (ManagementTaskProvider taskProvider : mTaskProviders) {
      BlockManagementTask task = taskProvider.getTask();
      if (task != null) {
        return task;
      }
    }
    // No task provided.
    return null;
  }

  /**
   * Main management loop.
   */
  private void runManagement() {
    while (true) {
      if (Thread.interrupted()) {
        // Coordinator closed.
        LOG.debug("Coordinator interrupted.");
        break;
      }

      BlockManagementTask currentTask;
      try {
        // Back off from worker if configured so.
        if (BACKOFF_STRATEGY == BackoffStrategy.ANY
            && mLoadTracker.loadDetected(BlockStoreLocation.anyTier())) {
          LOG.debug("Load detected. Sleeping {}ms.", LOAD_DETECTION_COOL_DOWN_TIME);
          Thread.sleep(LOAD_DETECTION_COOL_DOWN_TIME);
          continue;
        }

        final BlockManagementTask nextTask = getNextTask();
        if (nextTask == null) {
          LOG.debug("No management task pending. Sleeping {}ms.", LOAD_DETECTION_COOL_DOWN_TIME);
          Thread.sleep(LOAD_DETECTION_COOL_DOWN_TIME);
          continue;
        }

        // Submit and wait for the task.
        currentTask = nextTask;
        LOG.debug("Running task of type:{}", currentTask.getClass().getSimpleName());
        // Run the current task on coordinator thread.
        try {
          BlockManagementTaskResult result = currentTask.run();
          LOG.info("{} finished with result: {}", currentTask.getClass().getSimpleName(), result);

          if (result.noProgress()) {
            LOG.debug("Task made no progress due to failures/back-offs. Sleeping {}ms",
                LOAD_DETECTION_COOL_DOWN_TIME);
            Thread.sleep(LOAD_DETECTION_COOL_DOWN_TIME);
          }
        } catch (Exception e) {
          LOG.error("Management task failed: {}. Error: ", currentTask.getClass().getSimpleName(),
              e);
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        break;
      } catch (Throwable t) {
        LOG.error("Unexpected error during block management: ", t);
      }
    }
    LOG.debug("Block management coordinator exited.");
  }
}
