<commit_before>/*
 * Licensed to the University of California, Berkeley 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 tachyon.worker.block;

import java.io.IOException;
import java.util.List;
import java.util.Set;

import tachyon.Sessions;
import tachyon.client.UnderStorageType;
import tachyon.client.WorkerBlockMasterClient;
import tachyon.client.WorkerFileSystemMasterClient;
import tachyon.exception.BlockAlreadyExistsException;
import tachyon.exception.BlockDoesNotExistException;
import tachyon.exception.FailedToCheckpointException;
import tachyon.exception.InvalidWorkerStateException;
import tachyon.exception.TachyonException;
import tachyon.exception.WorkerOutOfSpaceException;
import tachyon.thrift.FileInfo;
import tachyon.thrift.TachyonTException;
import tachyon.underfs.UnderFileSystem;
import tachyon.util.io.FileUtils;
import tachyon.util.io.PathUtils;
import tachyon.worker.WorkerContext;
import tachyon.worker.WorkerIdRegistry;
import tachyon.worker.WorkerSource;
import tachyon.worker.block.io.BlockReader;
import tachyon.worker.block.io.BlockWriter;
import tachyon.worker.block.meta.BlockMeta;
import tachyon.worker.block.meta.TempBlockMeta;

/**
 * Class is responsible for managing the Tachyon BlockStore and Under FileSystem. This class is
 * thread-safe.
 */
public final class BlockDataManager {
  /** Block store delta reporter for master heartbeat */
  private BlockHeartbeatReporter mHeartbeatReporter;
  /** Block Store manager */
  private BlockStore mBlockStore;
  /** WorkerSource for collecting worker metrics */
  private WorkerSource mWorkerSource;
  /** Metrics reporter that listens on block events and increases metrics counters */
  private BlockMetricsReporter mMetricsReporter;

  /** WorkerBlockMasterClient, only used to inform the master of a new block in commitBlock */
  private WorkerBlockMasterClient mBlockMasterClient;
  /** WorkerFileSystemMasterClient, only used to inform master of a new file in persistFile */
  private WorkerFileSystemMasterClient mFileSystemMasterClient;
  /** Session metadata, used to keep track of session heartbeats */
  private Sessions mSessions = new Sessions();

  /**
   * Creates a BlockDataManager based on the configuration values.
   *
   * @param workerSource object for collecting the worker metrics
   * @param workerBlockMasterClient the block Tachyon master client for worker
   * @param workerFileSystemMasterClient the file system Tachyon master client for worker
   * @param blockStore the block store manager
   * @throws IOException if fail to connect to under filesystem
   */
  public BlockDataManager(WorkerSource workerSource,
      WorkerBlockMasterClient workerBlockMasterClient,
      WorkerFileSystemMasterClient workerFileSystemMasterClient, BlockStore blockStore)
          throws IOException {
    mHeartbeatReporter = new BlockHeartbeatReporter();
    mBlockStore = blockStore;
    mWorkerSource = workerSource;
    mMetricsReporter = new BlockMetricsReporter(mWorkerSource);

    mBlockMasterClient = workerBlockMasterClient;
    mFileSystemMasterClient = workerFileSystemMasterClient;

    // Register the heartbeat reporter so it can record block store changes
    mBlockStore.registerBlockStoreEventListener(mHeartbeatReporter);
    mBlockStore.registerBlockStoreEventListener(mMetricsReporter);
  }

  /**
   * Aborts the temporary block created by the session.
   *
   * @param sessionId The id of the client
   * @param blockId The id of the block to be aborted
   * @throws BlockAlreadyExistsException if blockId already exists in committed blocks
   * @throws BlockDoesNotExistException if the temporary block cannot be found
   * @throws InvalidWorkerStateException if blockId does not belong to sessionId
   * @throws IOException if temporary block cannot be deleted
   */
  public void abortBlock(long sessionId, long blockId) throws BlockAlreadyExistsException,
      BlockDoesNotExistException, InvalidWorkerStateException, IOException {
    mBlockStore.abortBlock(sessionId, blockId);
  }

  /**
   * Access the block for a given session. This should be called to update the evictor when
   * necessary.
   *
   * @param sessionId The id of the client
   * @param blockId The id of the block to access
   * @throws BlockDoesNotExistException this exception is not thrown in the tiered block store
   *         implementation
   */
  public void accessBlock(long sessionId, long blockId) throws BlockDoesNotExistException {
    mBlockStore.accessBlock(sessionId, blockId);
  }

  /**
   * Completes the process of persisting a file by renaming it to its final destination.
   *
   * This method is normally triggered from {@link tachyon.client.file.FileOutStream#close()} if and
   * only if {@link UnderStorageType#isSyncPersist()} ()} is true. The current implementation of
   * persistence is that through {@link tachyon.client.UnderStorageType} operations write to
   * {@link tachyon.underfs.UnderFileSystem} on the client's write path, but under a temporary file.
   *
   * @param fileId a file id
   * @param nonce a nonce used for temporary file creation
   * @param ufsPath the UFS path of the file
   * @throws TachyonTException if the file does not exist or cannot be renamed
   * @throws IOException if the update to the master fails
   */
  public void persistFile(long fileId, long nonce, String ufsPath)
      throws TachyonException, IOException {
    String tmpPath = PathUtils.temporaryFileName(fileId, nonce, ufsPath);
    UnderFileSystem ufs = UnderFileSystem.get(tmpPath, WorkerContext.getConf());
    try {
      if (!ufs.exists(tmpPath)) {
        // Location of the temporary file has changed, recompute it.
        FileInfo fileInfo = mFileSystemMasterClient.getFileInfo(fileId);
        ufsPath = fileInfo.getUfsPath();
        tmpPath = PathUtils.temporaryFileName(fileId, nonce, ufsPath);
      }
      if (!ufs.rename(tmpPath, ufsPath)) {
        throw new FailedToCheckpointException("Failed to rename " + tmpPath + " to " + ufsPath);
      }
    } catch (IOException ioe) {
      throw new FailedToCheckpointException(
          "Failed to rename " + tmpPath + " to " + ufsPath + ": " + ioe);
    }
    long fileSize;
    try {
      fileSize = ufs.getFileSize(ufsPath);
    } catch (IOException ioe) {
      throw new FailedToCheckpointException("Failed to getFileSize " + ufsPath);
    }
    mFileSystemMasterClient.persistFile(fileId, fileSize);<commit_msg>Fix the exception javadoc in BlockDataManager.<commit_after>/*
 * Licensed to the University of California, Berkeley 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 tachyon.worker.block;

import java.io.IOException;
import java.util.List;
import java.util.Set;

import tachyon.Sessions;
import tachyon.client.UnderStorageType;
import tachyon.client.WorkerBlockMasterClient;
import tachyon.client.WorkerFileSystemMasterClient;
import tachyon.exception.BlockAlreadyExistsException;
import tachyon.exception.BlockDoesNotExistException;
import tachyon.exception.FailedToCheckpointException;
import tachyon.exception.InvalidWorkerStateException;
import tachyon.exception.TachyonException;
import tachyon.exception.WorkerOutOfSpaceException;
import tachyon.thrift.FileInfo;
import tachyon.underfs.UnderFileSystem;
import tachyon.util.io.FileUtils;
import tachyon.util.io.PathUtils;
import tachyon.worker.WorkerContext;
import tachyon.worker.WorkerIdRegistry;
import tachyon.worker.WorkerSource;
import tachyon.worker.block.io.BlockReader;
import tachyon.worker.block.io.BlockWriter;
import tachyon.worker.block.meta.BlockMeta;
import tachyon.worker.block.meta.TempBlockMeta;

/**
 * Class is responsible for managing the Tachyon BlockStore and Under FileSystem. This class is
 * thread-safe.
 */
public final class BlockDataManager {
  /** Block store delta reporter for master heartbeat */
  private BlockHeartbeatReporter mHeartbeatReporter;
  /** Block Store manager */
  private BlockStore mBlockStore;
  /** WorkerSource for collecting worker metrics */
  private WorkerSource mWorkerSource;
  /** Metrics reporter that listens on block events and increases metrics counters */
  private BlockMetricsReporter mMetricsReporter;

  /** WorkerBlockMasterClient, only used to inform the master of a new block in commitBlock */
  private WorkerBlockMasterClient mBlockMasterClient;
  /** WorkerFileSystemMasterClient, only used to inform master of a new file in persistFile */
  private WorkerFileSystemMasterClient mFileSystemMasterClient;
  /** Session metadata, used to keep track of session heartbeats */
  private Sessions mSessions = new Sessions();

  /**
   * Creates a BlockDataManager based on the configuration values.
   *
   * @param workerSource object for collecting the worker metrics
   * @param workerBlockMasterClient the block Tachyon master client for worker
   * @param workerFileSystemMasterClient the file system Tachyon master client for worker
   * @param blockStore the block store manager
   * @throws IOException if fail to connect to under filesystem
   */
  public BlockDataManager(WorkerSource workerSource,
      WorkerBlockMasterClient workerBlockMasterClient,
      WorkerFileSystemMasterClient workerFileSystemMasterClient, BlockStore blockStore)
          throws IOException {
    mHeartbeatReporter = new BlockHeartbeatReporter();
    mBlockStore = blockStore;
    mWorkerSource = workerSource;
    mMetricsReporter = new BlockMetricsReporter(mWorkerSource);

    mBlockMasterClient = workerBlockMasterClient;
    mFileSystemMasterClient = workerFileSystemMasterClient;

    // Register the heartbeat reporter so it can record block store changes
    mBlockStore.registerBlockStoreEventListener(mHeartbeatReporter);
    mBlockStore.registerBlockStoreEventListener(mMetricsReporter);
  }

  /**
   * Aborts the temporary block created by the session.
   *
   * @param sessionId The id of the client
   * @param blockId The id of the block to be aborted
   * @throws BlockAlreadyExistsException if blockId already exists in committed blocks
   * @throws BlockDoesNotExistException if the temporary block cannot be found
   * @throws InvalidWorkerStateException if blockId does not belong to sessionId
   * @throws IOException if temporary block cannot be deleted
   */
  public void abortBlock(long sessionId, long blockId) throws BlockAlreadyExistsException,
      BlockDoesNotExistException, InvalidWorkerStateException, IOException {
    mBlockStore.abortBlock(sessionId, blockId);
  }

  /**
   * Access the block for a given session. This should be called to update the evictor when
   * necessary.
   *
   * @param sessionId The id of the client
   * @param blockId The id of the block to access
   * @throws BlockDoesNotExistException this exception is not thrown in the tiered block store
   *         implementation
   */
  public void accessBlock(long sessionId, long blockId) throws BlockDoesNotExistException {
    mBlockStore.accessBlock(sessionId, blockId);
  }

  /**
   * Completes the process of persisting a file by renaming it to its final destination.
   *
   * This method is normally triggered from {@link tachyon.client.file.FileOutStream#close()} if and
   * only if {@link UnderStorageType#isSyncPersist()} ()} is true. The current implementation of
   * persistence is that through {@link tachyon.client.UnderStorageType} operations write to
   * {@link tachyon.underfs.UnderFileSystem} on the client's write path, but under a temporary file.
   *
   * @param fileId a file id
   * @param nonce a nonce used for temporary file creation
   * @param ufsPath the UFS path of the file
   * @throws IOException if the update to the master fails
   * @throws TachyonException if the file does not exist or cannot be renamed
   */
  public void persistFile(long fileId, long nonce, String ufsPath)
      throws IOException, TachyonException {
    String tmpPath = PathUtils.temporaryFileName(fileId, nonce, ufsPath);
    UnderFileSystem ufs = UnderFileSystem.get(tmpPath, WorkerContext.getConf());
    try {
      if (!ufs.exists(tmpPath)) {
        // Location of the temporary file has changed, recompute it.
        FileInfo fileInfo = mFileSystemMasterClient.getFileInfo(fileId);
        ufsPath = fileInfo.getUfsPath();
        tmpPath = PathUtils.temporaryFileName(fileId, nonce, ufsPath);
      }
      if (!ufs.rename(tmpPath, ufsPath)) {
        throw new FailedToCheckpointException("Failed to rename " + tmpPath + " to " + ufsPath);
      }
    } catch (IOException ioe) {
      throw new FailedToCheckpointException(
          "Failed to rename " + tmpPath + " to " + ufsPath + ": " + ioe);
    }
    long fileSize;
    try {
      fileSize = ufs.getFileSize(ufsPath);
    } catch (IOException ioe) {
      throw new FailedToCheckpointException("Failed to getFileSize " + ufsPath);
    }
    mFileSystemMasterClient.persistFile(fileId, fileSize);
{'fraction_non_alphanumeric': 0.06626802673232501, 'fraction_numerical': 0.0005627857896588111, 'mean_word_length': 4.318368873924429, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1693303', 'n_tokens_mistral': 3840, 'n_tokens_neox': 3722, 'n_words': 1334}