Java 类org.apache.hadoop.hdfs.server.namenode.INodesInPath 实例源码

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }

  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    d.setSnapshotQuota(DirectorySnapshottableFeature.SNAPSHOT_LIMIT);
  } else {
    d.addSnapshottableFeature();
  }
  addSnapshottable(d);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  DirectorySnapshottableFeature sf = d.getDirectorySnapshottableFeature();
  if (sf == null) {
    // the directory is already non-snapshottable
    return;
  }
  if (sf.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (d == fsdir.getRoot()) {
    d.setSnapshotQuota(0);
  } else {
    d.removeSnapshottableFeature();
  }
  removeSnapshottable(d);
}

/**
 * Create a snapshot of the given path.
 * It is assumed that the caller will perform synchronization.
 *
 * @param iip the INodes resolved from the snapshottable directory's path
 * @param snapshotName
 *          The name of the snapshot.
 * @throws IOException
 *           Throw IOException when 1) the given path does not lead to an
 *           existing snapshottable directory, and/or 2) there exists a
 *           snapshot with the given name for the directory, and/or 3)
 *           snapshot number exceeds quota
 */
public String createSnapshot(final INodesInPath iip, String snapshotRoot,
    String snapshotName) throws IOException {
  INodeDirectory srcRoot = getSnapshottableRoot(iip);

  if (snapshotCounter == getMaxSnapshotID()) {
    // We have reached the maximum allowable snapshot ID and since we don't
    // handle rollover we will fail all subsequent snapshot creation
    // requests.
    //
    throw new SnapshotException(
        "Failed to create the snapshot. The FileSystem has run out of " +
        "snapshot IDs and ID rollover is not supported.");
  }

  srcRoot.addSnapshot(snapshotCounter, snapshotName);

  //create success, update id
  snapshotCounter++;
  numSnapshots.getAndIncrement();
  return Snapshot.getSnapshotPath(snapshotRoot, snapshotName);
}

/**
 * Compute the difference between two snapshots of a directory, or between a
 * snapshot of the directory and its current tree.
 */
public SnapshotDiffReport diff(final INodesInPath iip,
    final String snapshotRootPath, final String from,
    final String to) throws IOException {
  // Find the source root directory path where the snapshots were taken.
  // All the check for path has been included in the valueOf method.
  final INodeDirectory snapshotRoot = getSnapshottableRoot(iip);

  if ((from == null || from.isEmpty())
      && (to == null || to.isEmpty())) {
    // both fromSnapshot and toSnapshot indicate the current tree
    return new SnapshotDiffReport(snapshotRootPath, from, to,
        Collections.<DiffReportEntry> emptyList());
  }
  final SnapshotDiffInfo diffs = snapshotRoot
      .getDirectorySnapshottableFeature().computeDiff(snapshotRoot, from, to);
  return diffs != null ? diffs.generateReport() : new SnapshotDiffReport(
      snapshotRootPath, from, to, Collections.<DiffReportEntry> emptyList());
}

/**
 * Check the replication for both the current file and all its prior snapshots
 * 
 * @param currentFile
 *          the Path of the current file
 * @param snapshotRepMap
 *          A map maintaining all the snapshots of the current file, as well
 *          as their expected replication number stored in their corresponding
 *          INodes
 * @param expectedBlockRep
 *          The expected replication number
 * @throws Exception
 */
private void checkSnapshotFileReplication(Path currentFile,
    Map<Path, Short> snapshotRepMap, short expectedBlockRep) throws Exception {
  // First check the getBlockReplication for the INode of the currentFile
  final INodeFile inodeOfCurrentFile = getINodeFile(currentFile);
  assertEquals(expectedBlockRep, inodeOfCurrentFile.getBlockReplication());
  // Then check replication for every snapshot
  for (Path ss : snapshotRepMap.keySet()) {
    final INodesInPath iip = fsdir.getINodesInPath(ss.toString(), true);
    final INodeFile ssInode = iip.getLastINode().asFile();
    // The replication number derived from the
    // INodeFileWithLink#getBlockReplication should always == expectedBlockRep
    assertEquals(expectedBlockRep, ssInode.getBlockReplication());
    // Also check the number derived from INodeFile#getFileReplication
    assertEquals(snapshotRepMap.get(ss).shortValue(),
        ssInode.getFileReplication(iip.getPathSnapshotId()));
  }
}

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }

  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    d.setSnapshotQuota(DirectorySnapshottableFeature.SNAPSHOT_LIMIT);
  } else {
    d.addSnapshottableFeature();
  }
  addSnapshottable(d);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  DirectorySnapshottableFeature sf = d.getDirectorySnapshottableFeature();
  if (sf == null) {
    // the directory is already non-snapshottable
    return;
  }
  if (sf.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (d == fsdir.getRoot()) {
    d.setSnapshotQuota(0);
  } else {
    d.removeSnapshottableFeature();
  }
  removeSnapshottable(d);
}

/**
 * Create a snapshot of the given path.
 * It is assumed that the caller will perform synchronization.
 *
 * @param iip the INodes resolved from the snapshottable directory's path
 * @param snapshotName
 *          The name of the snapshot.
 * @throws IOException
 *           Throw IOException when 1) the given path does not lead to an
 *           existing snapshottable directory, and/or 2) there exists a
 *           snapshot with the given name for the directory, and/or 3)
 *           snapshot number exceeds quota
 */
public String createSnapshot(final INodesInPath iip, String snapshotRoot,
    String snapshotName) throws IOException {
  INodeDirectory srcRoot = getSnapshottableRoot(iip);

  if (snapshotCounter == getMaxSnapshotID()) {
    // We have reached the maximum allowable snapshot ID and since we don't
    // handle rollover we will fail all subsequent snapshot creation
    // requests.
    throw new SnapshotException(
        "Failed to create the snapshot. The FileSystem has run out of " +
        "snapshot IDs and ID rollover is not supported.");
  }

  srcRoot.addSnapshot(snapshotCounter, snapshotName);

  //create success, update id
  snapshotCounter++;
  numSnapshots.getAndIncrement();
  return Snapshot.getSnapshotPath(snapshotRoot, snapshotName);
}

/**
 * Compute the difference between two snapshots of a directory, or between a
 * snapshot of the directory and its current tree.
 */
public SnapshotDiffReport diff(final INodesInPath iip,
    final String snapshotRootPath, final String from,
    final String to) throws IOException {
  // Find the source root directory path where the snapshots were taken.
  // All the check for path has been included in the valueOf method.
  final INodeDirectory snapshotRoot = getSnapshottableRoot(iip);

  if ((from == null || from.isEmpty())
      && (to == null || to.isEmpty())) {
    // both fromSnapshot and toSnapshot indicate the current tree
    return new SnapshotDiffReport(snapshotRootPath, from, to,
        Collections.<DiffReportEntry> emptyList());
  }
  final SnapshotDiffInfo diffs = snapshotRoot
      .getDirectorySnapshottableFeature().computeDiff(snapshotRoot, from, to);
  return diffs != null ? diffs.generateReport() : new SnapshotDiffReport(
      snapshotRootPath, from, to, Collections.<DiffReportEntry> emptyList());
}

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }

  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    d.setSnapshotQuota(DirectorySnapshottableFeature.SNAPSHOT_LIMIT);
  } else {
    d.addSnapshottableFeature();
  }
  addSnapshottable(d);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  DirectorySnapshottableFeature sf = d.getDirectorySnapshottableFeature();
  if (sf == null) {
    // the directory is already non-snapshottable
    return;
  }
  if (sf.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (d == fsdir.getRoot()) {
    d.setSnapshotQuota(0);
  } else {
    d.removeSnapshottableFeature();
  }
  removeSnapshottable(d);
}

/**
 * Create a snapshot of the given path.
 * It is assumed that the caller will perform synchronization.
 *
 * @param iip the INodes resolved from the snapshottable directory's path
 * @param snapshotName
 *          The name of the snapshot.
 * @throws IOException
 *           Throw IOException when 1) the given path does not lead to an
 *           existing snapshottable directory, and/or 2) there exists a
 *           snapshot with the given name for the directory, and/or 3)
 *           snapshot number exceeds quota
 */
public String createSnapshot(final INodesInPath iip, String snapshotRoot,
    String snapshotName) throws IOException {
  INodeDirectory srcRoot = getSnapshottableRoot(iip);

  if (snapshotCounter == getMaxSnapshotID()) {
    // We have reached the maximum allowable snapshot ID and since we don't
    // handle rollover we will fail all subsequent snapshot creation
    // requests.
    //
    throw new SnapshotException(
        "Failed to create the snapshot. The FileSystem has run out of " +
        "snapshot IDs and ID rollover is not supported.");
  }

  srcRoot.addSnapshot(snapshotCounter, snapshotName);

  //create success, update id
  snapshotCounter++;
  numSnapshots.getAndIncrement();
  return Snapshot.getSnapshotPath(snapshotRoot, snapshotName);
}

/**
 * Compute the difference between two snapshots of a directory, or between a
 * snapshot of the directory and its current tree.
 */
public SnapshotDiffReport diff(final INodesInPath iip,
    final String snapshotRootPath, final String from,
    final String to) throws IOException {
  // Find the source root directory path where the snapshots were taken.
  // All the check for path has been included in the valueOf method.
  final INodeDirectory snapshotRoot = getSnapshottableRoot(iip);

  if ((from == null || from.isEmpty())
      && (to == null || to.isEmpty())) {
    // both fromSnapshot and toSnapshot indicate the current tree
    return new SnapshotDiffReport(snapshotRootPath, from, to,
        Collections.<DiffReportEntry> emptyList());
  }
  final SnapshotDiffInfo diffs = snapshotRoot
      .getDirectorySnapshottableFeature().computeDiff(snapshotRoot, from, to);
  return diffs != null ? diffs.generateReport() : new SnapshotDiffReport(
      snapshotRootPath, from, to, Collections.<DiffReportEntry> emptyList());
}

/**
 * Check the replication for both the current file and all its prior snapshots
 * 
 * @param currentFile
 *          the Path of the current file
 * @param snapshotRepMap
 *          A map maintaining all the snapshots of the current file, as well
 *          as their expected replication number stored in their corresponding
 *          INodes
 * @param expectedBlockRep
 *          The expected replication number
 * @throws Exception
 */
private void checkSnapshotFileReplication(Path currentFile,
    Map<Path, Short> snapshotRepMap, short expectedBlockRep) throws Exception {
  // First check the getBlockReplication for the INode of the currentFile
  final INodeFile inodeOfCurrentFile = getINodeFile(currentFile);
  assertEquals(expectedBlockRep, inodeOfCurrentFile.getBlockReplication());
  // Then check replication for every snapshot
  for (Path ss : snapshotRepMap.keySet()) {
    final INodesInPath iip = fsdir.getINodesInPath(ss.toString(), true);
    final INodeFile ssInode = iip.getLastINode().asFile();
    // The replication number derived from the
    // INodeFileWithLink#getBlockReplication should always == expectedBlockRep
    assertEquals(expectedBlockRep, ssInode.getBlockReplication());
    // Also check the number derived from INodeFile#getFileReplication
    assertEquals(snapshotRepMap.get(ss).shortValue(),
        ssInode.getFileReplication(iip.getPathSnapshotId()));
  }
}

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }

  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    d.setSnapshotQuota(DirectorySnapshottableFeature.SNAPSHOT_LIMIT);
  } else {
    d.addSnapshottableFeature();
  }
  addSnapshottable(d);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  DirectorySnapshottableFeature sf = d.getDirectorySnapshottableFeature();
  if (sf == null) {
    // the directory is already non-snapshottable
    return;
  }
  if (sf.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (d == fsdir.getRoot()) {
    d.setSnapshotQuota(0);
  } else {
    d.removeSnapshottableFeature();
  }
  removeSnapshottable(d);
}

/**
 * Check the replication for both the current file and all its prior snapshots
 * 
 * @param currentFile
 *          the Path of the current file
 * @param snapshotRepMap
 *          A map maintaining all the snapshots of the current file, as well
 *          as their expected replication number stored in their corresponding
 *          INodes
 * @param expectedBlockRep
 *          The expected replication number that should be returned by
 *          {@link INodeFileWithSnapshot#getBlockReplication()} of all the INodes
 * @throws Exception
 */
private void checkSnapshotFileReplication(Path currentFile,
    Map<Path, Short> snapshotRepMap, short expectedBlockRep) throws Exception {
  // First check the getBlockReplication for the INode of the currentFile
  final INodeFile inodeOfCurrentFile = getINodeFile(currentFile);
  assertEquals(expectedBlockRep, inodeOfCurrentFile.getBlockReplication());
  // Then check replication for every snapshot
  for (Path ss : snapshotRepMap.keySet()) {
    final INodesInPath iip = fsdir.getLastINodeInPath(ss.toString());
    final INodeFile ssInode = (INodeFile)iip.getLastINode();
    // The replication number derived from the
    // INodeFileWithLink#getBlockReplication should always == expectedBlockRep
    assertEquals(expectedBlockRep, ssInode.getBlockReplication());
    // Also check the number derived from INodeFile#getFileReplication
    assertEquals(snapshotRepMap.get(ss).shortValue(),
        ssInode.getFileReplication(iip.getPathSnapshotId()));
  }
}

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }


  final INodeDirectorySnapshottable s;
  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    s = (INodeDirectorySnapshottable)d; 
    s.setSnapshotQuota(INodeDirectorySnapshottable.SNAPSHOT_LIMIT);
  } else {
    s = d.replaceSelf4INodeDirectorySnapshottable(iip.getLatestSnapshot(),
        fsdir.getINodeMap());
  }
  addSnapshottable(s);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (!d.isSnapshottable()) {
    // the directory is already non-snapshottable
    return;
  }
  final INodeDirectorySnapshottable s = (INodeDirectorySnapshottable) d;
  if (s.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (s == fsdir.getRoot()) {
    s.setSnapshotQuota(0); 
  } else {
    s.replaceSelf(iip.getLatestSnapshot(), fsdir.getINodeMap());
  }
  removeSnapshottable(s);
}

/**
 * Compute the difference between two snapshots of a directory, or between a
 * snapshot of the directory and its current tree.
 */
public SnapshotDiffInfo diff(final String path, final String from,
    final String to) throws IOException {
  if ((from == null || from.isEmpty())
      && (to == null || to.isEmpty())) {
    // both fromSnapshot and toSnapshot indicate the current tree
    return null;
  }

  // Find the source root directory path where the snapshots were taken.
  // All the check for path has been included in the valueOf method.
  INodesInPath inodesInPath = fsdir.getINodesInPath4Write(path.toString());
  final INodeDirectorySnapshottable snapshotRoot = INodeDirectorySnapshottable
      .valueOf(inodesInPath.getLastINode(), path);

  return snapshotRoot.computeDiff(from, to);
}

/**
 * Check the replication for both the current file and all its prior snapshots
 * 
 * @param currentFile
 *          the Path of the current file
 * @param snapshotRepMap
 *          A map maintaining all the snapshots of the current file, as well
 *          as their expected replication number stored in their corresponding
 *          INodes
 * @param expectedBlockRep
 *          The expected replication number that should be returned by
 *          {@link INodeFileWithSnapshot#getBlockReplication()} of all the INodes
 * @throws Exception
 */
private void checkSnapshotFileReplication(Path currentFile,
    Map<Path, Short> snapshotRepMap, short expectedBlockRep) throws Exception {
  // First check the getBlockReplication for the INode of the currentFile
  final INodeFile inodeOfCurrentFile = getINodeFile(currentFile);
  assertEquals(expectedBlockRep, inodeOfCurrentFile.getBlockReplication());
  // Then check replication for every snapshot
  for (Path ss : snapshotRepMap.keySet()) {
    final INodesInPath iip = fsdir.getLastINodeInPath(ss.toString());
    final INodeFile ssInode = (INodeFile)iip.getLastINode();
    // The replication number derived from the
    // INodeFileWithLink#getBlockReplication should always == expectedBlockRep
    assertEquals(expectedBlockRep, ssInode.getBlockReplication());
    // Also check the number derived from INodeFile#getFileReplication
    assertEquals(snapshotRepMap.get(ss).shortValue(),
        ssInode.getFileReplication(iip.getPathSnapshot()));
  }
}

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }

  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    d.setSnapshotQuota(DirectorySnapshottableFeature.SNAPSHOT_LIMIT);
  } else {
    d.addSnapshottableFeature();
  }
  addSnapshottable(d);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  DirectorySnapshottableFeature sf = d.getDirectorySnapshottableFeature();
  if (sf == null) {
    // the directory is already non-snapshottable
    return;
  }
  if (sf.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (d == fsdir.getRoot()) {
    d.setSnapshotQuota(0);
  } else {
    d.removeSnapshottableFeature();
  }
  removeSnapshottable(d);
}

/**
 * Check the replication for both the current file and all its prior snapshots
 * 
 * @param currentFile
 *          the Path of the current file
 * @param snapshotRepMap
 *          A map maintaining all the snapshots of the current file, as well
 *          as their expected replication number stored in their corresponding
 *          INodes
 * @param expectedBlockRep
 *          The expected replication number that should be returned by
 *          {@link INodeFileWithSnapshot#getBlockReplication()} of all the INodes
 * @throws Exception
 */
private void checkSnapshotFileReplication(Path currentFile,
    Map<Path, Short> snapshotRepMap, short expectedBlockRep) throws Exception {
  // First check the getBlockReplication for the INode of the currentFile
  final INodeFile inodeOfCurrentFile = getINodeFile(currentFile);
  assertEquals(expectedBlockRep, inodeOfCurrentFile.getBlockReplication());
  // Then check replication for every snapshot
  for (Path ss : snapshotRepMap.keySet()) {
    final INodesInPath iip = fsdir.getLastINodeInPath(ss.toString());
    final INodeFile ssInode = (INodeFile)iip.getLastINode();
    // The replication number derived from the
    // INodeFileWithLink#getBlockReplication should always == expectedBlockRep
    assertEquals(expectedBlockRep, ssInode.getBlockReplication());
    // Also check the number derived from INodeFile#getFileReplication
    assertEquals(snapshotRepMap.get(ss).shortValue(),
        ssInode.getFileReplication(iip.getPathSnapshotId()));
  }
}

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }


  final INodeDirectorySnapshottable s;
  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    s = (INodeDirectorySnapshottable)d; 
    s.setSnapshotQuota(INodeDirectorySnapshottable.SNAPSHOT_LIMIT);
  } else {
    s = d.replaceSelf4INodeDirectorySnapshottable(iip.getLatestSnapshot(),
        fsdir.getINodeMap());
  }
  addSnapshottable(s);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (!d.isSnapshottable()) {
    // the directory is already non-snapshottable
    return;
  }
  final INodeDirectorySnapshottable s = (INodeDirectorySnapshottable) d;
  if (s.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (s == fsdir.getRoot()) {
    s.setSnapshotQuota(0); 
  } else {
    s.replaceSelf(iip.getLatestSnapshot(), fsdir.getINodeMap());
  }
  removeSnapshottable(s);
}

/**
 * Compute the difference between two snapshots of a directory, or between a
 * snapshot of the directory and its current tree.
 */
public SnapshotDiffInfo diff(final String path, final String from,
    final String to) throws IOException {
  if ((from == null || from.isEmpty())
      && (to == null || to.isEmpty())) {
    // both fromSnapshot and toSnapshot indicate the current tree
    return null;
  }

  // Find the source root directory path where the snapshots were taken.
  // All the check for path has been included in the valueOf method.
  INodesInPath inodesInPath = fsdir.getINodesInPath4Write(path.toString());
  final INodeDirectorySnapshottable snapshotRoot = INodeDirectorySnapshottable
      .valueOf(inodesInPath.getLastINode(), path);

  return snapshotRoot.computeDiff(from, to);
}

/**
 * Check the replication for both the current file and all its prior snapshots
 * 
 * @param currentFile
 *          the Path of the current file
 * @param snapshotRepMap
 *          A map maintaining all the snapshots of the current file, as well
 *          as their expected replication number stored in their corresponding
 *          INodes
 * @param expectedBlockRep
 *          The expected replication number that should be returned by
 *          {@link INodeFileWithSnapshot#getBlockReplication()} of all the INodes
 * @throws Exception
 */
private void checkSnapshotFileReplication(Path currentFile,
    Map<Path, Short> snapshotRepMap, short expectedBlockRep) throws Exception {
  // First check the getBlockReplication for the INode of the currentFile
  final INodeFile inodeOfCurrentFile = getINodeFile(currentFile);
  assertEquals(expectedBlockRep, inodeOfCurrentFile.getBlockReplication());
  // Then check replication for every snapshot
  for (Path ss : snapshotRepMap.keySet()) {
    final INodesInPath iip = fsdir.getLastINodeInPath(ss.toString());
    final INodeFile ssInode = (INodeFile)iip.getLastINode();
    // The replication number derived from the
    // INodeFileWithLink#getBlockReplication should always == expectedBlockRep
    assertEquals(expectedBlockRep, ssInode.getBlockReplication());
    // Also check the number derived from INodeFile#getFileReplication
    assertEquals(snapshotRepMap.get(ss).shortValue(),
        ssInode.getFileReplication(iip.getPathSnapshot()));
  }
}

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }


  final INodeDirectorySnapshottable s;
  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    s = (INodeDirectorySnapshottable)d; 
    s.setSnapshotQuota(INodeDirectorySnapshottable.SNAPSHOT_LIMIT);
  } else {
    s = d.replaceSelf4INodeDirectorySnapshottable(iip.getLatestSnapshot(),
        fsdir.getINodeMap());
  }
  addSnapshottable(s);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (!d.isSnapshottable()) {
    // the directory is already non-snapshottable
    return;
  }
  final INodeDirectorySnapshottable s = (INodeDirectorySnapshottable) d;
  if (s.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (s == fsdir.getRoot()) {
    s.setSnapshotQuota(0); 
  } else {
    s.replaceSelf(iip.getLatestSnapshot(), fsdir.getINodeMap());
  }
  removeSnapshottable(s);
}

/**
 * Compute the difference between two snapshots of a directory, or between a
 * snapshot of the directory and its current tree.
 */
public SnapshotDiffInfo diff(final String path, final String from,
    final String to) throws IOException {
  if ((from == null || from.isEmpty())
      && (to == null || to.isEmpty())) {
    // both fromSnapshot and toSnapshot indicate the current tree
    return null;
  }

  // Find the source root directory path where the snapshots were taken.
  // All the check for path has been included in the valueOf method.
  INodesInPath inodesInPath = fsdir.getINodesInPath4Write(path.toString());
  final INodeDirectorySnapshottable snapshotRoot = INodeDirectorySnapshottable
      .valueOf(inodesInPath.getLastINode(), path);

  return snapshotRoot.computeDiff(from, to);
}

/**
 * Check the replication for both the current file and all its prior snapshots
 * 
 * @param currentFile
 *          the Path of the current file
 * @param snapshotRepMap
 *          A map maintaining all the snapshots of the current file, as well
 *          as their expected replication number stored in their corresponding
 *          INodes
 * @param expectedBlockRep
 *          The expected replication number that should be returned by
 *          {@link INodeFileWithSnapshot#getBlockReplication()} of all the INodes
 * @throws Exception
 */
private void checkSnapshotFileReplication(Path currentFile,
    Map<Path, Short> snapshotRepMap, short expectedBlockRep) throws Exception {
  // First check the getBlockReplication for the INode of the currentFile
  final INodeFile inodeOfCurrentFile = getINodeFile(currentFile);
  assertEquals(expectedBlockRep, inodeOfCurrentFile.getBlockReplication());
  // Then check replication for every snapshot
  for (Path ss : snapshotRepMap.keySet()) {
    final INodesInPath iip = fsdir.getLastINodeInPath(ss.toString());
    final INodeFile ssInode = (INodeFile)iip.getLastINode();
    // The replication number derived from the
    // INodeFileWithLink#getBlockReplication should always == expectedBlockRep
    assertEquals(expectedBlockRep, ssInode.getBlockReplication());
    // Also check the number derived from INodeFile#getFileReplication
    assertEquals(snapshotRepMap.get(ss).shortValue(),
        ssInode.getFileReplication(iip.getPathSnapshot()));
  }
}

/**
 * Set the given directory as a snapshottable directory.
 * If the path is already a snapshottable directory, update the quota.
 */
public void setSnapshottable(final String path, boolean checkNestedSnapshottable)
    throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (checkNestedSnapshottable) {
    checkNestedSnapshottable(d, path);
  }


  final INodeDirectorySnapshottable s;
  if (d.isSnapshottable()) {
    //The directory is already a snapshottable directory.
    s = (INodeDirectorySnapshottable)d; 
    s.setSnapshotQuota(INodeDirectorySnapshottable.SNAPSHOT_LIMIT);
  } else {
    s = d.replaceSelf4INodeDirectorySnapshottable(iip.getLatestSnapshotId(),
        fsdir.getINodeMap());
  }
  addSnapshottable(s);
}

/**
 * Set the given snapshottable directory to non-snapshottable.
 * 
 * @throws SnapshotException if there are snapshots in the directory.
 */
public void resetSnapshottable(final String path) throws IOException {
  final INodesInPath iip = fsdir.getINodesInPath4Write(path);
  final INodeDirectory d = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (!d.isSnapshottable()) {
    // the directory is already non-snapshottable
    return;
  }
  final INodeDirectorySnapshottable s = (INodeDirectorySnapshottable) d;
  if (s.getNumSnapshots() > 0) {
    throw new SnapshotException("The directory " + path + " has snapshot(s). "
        + "Please redo the operation after removing all the snapshots.");
  }

  if (s == fsdir.getRoot()) {
    s.setSnapshotQuota(0); 
  } else {
    s.replaceSelf(iip.getLatestSnapshotId(), fsdir.getINodeMap());
  }
  removeSnapshottable(s);
}

/**
 * Compute the difference between two snapshots of a directory, or between a
 * snapshot of the directory and its current tree.
 */
public SnapshotDiffInfo diff(final String path, final String from,
    final String to) throws IOException {
  if ((from == null || from.isEmpty())
      && (to == null || to.isEmpty())) {
    // both fromSnapshot and toSnapshot indicate the current tree
    return null;
  }

  // Find the source root directory path where the snapshots were taken.
  // All the check for path has been included in the valueOf method.
  INodesInPath inodesInPath = fsdir.getINodesInPath4Write(path.toString());
  final INodeDirectorySnapshottable snapshotRoot = INodeDirectorySnapshottable
      .valueOf(inodesInPath.getLastINode(), path);

  return snapshotRoot.computeDiff(from, to);
}

/**
 * Check the replication for both the current file and all its prior snapshots
 * 
 * @param currentFile
 *          the Path of the current file
 * @param snapshotRepMap
 *          A map maintaining all the snapshots of the current file, as well
 *          as their expected replication number stored in their corresponding
 *          INodes
 * @param expectedBlockRep
 *          The expected replication number that should be returned by
 *          {@link INodeFileWithSnapshot#getBlockReplication()} of all the INodes
 * @throws Exception
 */
private void checkSnapshotFileReplication(Path currentFile,
    Map<Path, Short> snapshotRepMap, short expectedBlockRep) throws Exception {
  // First check the getBlockReplication for the INode of the currentFile
  final INodeFile inodeOfCurrentFile = getINodeFile(currentFile);
  assertEquals(expectedBlockRep, inodeOfCurrentFile.getBlockReplication());
  // Then check replication for every snapshot
  for (Path ss : snapshotRepMap.keySet()) {
    final INodesInPath iip = fsdir.getLastINodeInPath(ss.toString());
    final INodeFile ssInode = (INodeFile)iip.getLastINode();
    // The replication number derived from the
    // INodeFileWithLink#getBlockReplication should always == expectedBlockRep
    assertEquals(expectedBlockRep, ssInode.getBlockReplication());
    // Also check the number derived from INodeFile#getFileReplication
    assertEquals(snapshotRepMap.get(ss).shortValue(),
        ssInode.getFileReplication(iip.getPathSnapshotId()));
  }
}

/**
* Find the source root directory where the snapshot will be taken
* for a given path.
*
* @return Snapshottable directory.
* @throws IOException
*           Throw IOException when the given path does not lead to an
*           existing snapshottable directory.
*/
public INodeDirectory getSnapshottableRoot(final INodesInPath iip)
    throws IOException {
  final String path = iip.getPath();
  final INodeDirectory dir = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (!dir.isSnapshottable()) {
    throw new SnapshotException(
        "Directory is not a snapshottable directory: " + path);
  }
  return dir;
}

/**
 * Delete a snapshot for a snapshottable directory
 * @param snapshotName Name of the snapshot to be deleted
 * @param collectedBlocks Used to collect information to update blocksMap 
 * @throws IOException
 */
public void deleteSnapshot(final INodesInPath iip, final String snapshotName,
    BlocksMapUpdateInfo collectedBlocks, final List<INode> removedINodes)
    throws IOException {
  INodeDirectory srcRoot = getSnapshottableRoot(iip);
  srcRoot.removeSnapshot(fsdir.getBlockStoragePolicySuite(), snapshotName,
      collectedBlocks, removedINodes);
  numSnapshots.getAndDecrement();
}

/**
* Find the source root directory where the snapshot will be taken
* for a given path.
*
* @return Snapshottable directory.
* @throws IOException
*           Throw IOException when the given path does not lead to an
*           existing snapshottable directory.
*/
public INodeDirectory getSnapshottableRoot(final INodesInPath iip)
    throws IOException {
  final String path = iip.getPath();
  final INodeDirectory dir = INodeDirectory.valueOf(iip.getLastINode(), path);
  if (!dir.isSnapshottable()) {
    throw new SnapshotException(
        "Directory is not a snapshottable directory: " + path);
  }
  return dir;
}

/**
 * Delete a snapshot for a snapshottable directory
 * @param snapshotName Name of the snapshot to be deleted
 * @param reclaimContext Used to collect information to reclaim blocks
 *                       and inodes
 */
public void deleteSnapshot(final INodesInPath iip, final String snapshotName,
    INode.ReclaimContext reclaimContext) throws IOException {
  INodeDirectory srcRoot = getSnapshottableRoot(iip);
  srcRoot.removeSnapshot(reclaimContext, snapshotName);
  numSnapshots.getAndDecrement();
}