import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
ScanInfo(long blockId, File blockFile, File metaFile, FsVolumeSpi vol) {
  this.blockId = blockId;
  String condensedVolPath = vol == null ? null :
    getCondensedPath(vol.getBasePath());
  this.blockSuffix = blockFile == null ? null :
    getSuffix(blockFile, condensedVolPath);
  this.blockFileLength = (blockFile != null) ? blockFile.length() : 0; 
  if (metaFile == null) {
    this.metaSuffix = null;
  } else if (blockFile == null) {
    this.metaSuffix = getSuffix(metaFile, condensedVolPath);
  } else {
    this.metaSuffix = getSuffix(metaFile,
        condensedVolPath + blockSuffix);
  }
  this.volume = vol;
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
 * Stops and removes a volume scanner.<p/>
 *
 * This function will block until the volume scanner has stopped.
 *
 * @param volume           The volume to remove.
 */
public synchronized void removeVolumeScanner(FsVolumeSpi volume) {
  if (!isEnabled()) {
    LOG.debug("Not removing volume scanner for {}, because the block " +
        "scanner is disabled.", volume.getStorageID());
    return;
  }
  VolumeScanner scanner = scanners.get(volume.getStorageID());
  if (scanner == null) {
    LOG.warn("No scanner found to remove for volumeId {}",
        volume.getStorageID());
    return;
  }
  LOG.info("Removing scanner for volume {} (StorageID {})",
      volume.getBasePath(), volume.getStorageID());
  scanner.shutdown();
  scanners.remove(volume.getStorageID());
  Uninterruptibles.joinUninterruptibly(scanner, 5, TimeUnit.MINUTES);
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
 * Check that the permissions of the local DN directories are as expected.
 */
@Test
public void testLocalDirs() throws Exception {
  Configuration conf = new Configuration();
  final String permStr = conf.get(
    DFSConfigKeys.DFS_DATANODE_DATA_DIR_PERMISSION_KEY);
  FsPermission expected = new FsPermission(permStr);

  // Check permissions on directories in 'dfs.datanode.data.dir'
  FileSystem localFS = FileSystem.getLocal(conf);
  for (DataNode dn : cluster.getDataNodes()) {
    for (FsVolumeSpi v : dn.getFSDataset().getVolumes()) {
      String dir = v.getBasePath();
      Path dataDir = new Path(dir);
      FsPermission actual = localFS.getFileStatus(dataDir).getPermission();
        assertEquals("Permission for dir: " + dataDir + ", is " + actual +
            ", while expected is " + expected, expected, actual);
    }
  }
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
 * Create a ScanInfo object for a block. This constructor will examine
 * the block data and meta-data files.
 *
 * @param blockId the block ID
 * @param blockFile the path to the block data file
 * @param metaFile the path to the block meta-data file
 * @param vol the volume that contains the block
 */
ScanInfo(long blockId, File blockFile, File metaFile, FsVolumeSpi vol) {
  this.blockId = blockId;
  String condensedVolPath = vol == null ? null :
    getCondensedPath(vol.getBasePath());
  this.blockSuffix = blockFile == null ? null :
    getSuffix(blockFile, condensedVolPath);
  this.blockFileLength = (blockFile != null) ? blockFile.length() : 0; 
  if (metaFile == null) {
    this.metaSuffix = null;
  } else if (blockFile == null) {
    this.metaSuffix = getSuffix(metaFile, condensedVolPath);
  } else {
    this.metaSuffix = getSuffix(metaFile,
        condensedVolPath + blockSuffix);
  }
  this.volume = vol;
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
@Override
public Iterator<Replica> getStoredReplicas(String bpid) throws IOException {
  // Reload replicas from the disk.
  ReplicaMap replicaMap = new ReplicaMap(dataset);
  try (FsVolumeReferences refs = dataset.getFsVolumeReferences()) {
    for (FsVolumeSpi vol : refs) {
      FsVolumeImpl volume = (FsVolumeImpl) vol;
      volume.getVolumeMap(bpid, replicaMap, dataset.ramDiskReplicaTracker);
    }
  }

  // Cast ReplicaInfo to Replica, because ReplicaInfo assumes a file-based
  // FsVolumeSpi implementation.
  List<Replica> ret = new ArrayList<>();
  if (replicaMap.replicas(bpid) != null) {
    ret.addAll(replicaMap.replicas(bpid));
  }
  return ret.iterator();
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
private void createReplicas(List<String> bpList, List<FsVolumeSpi> volumes,
                            FsDatasetTestUtils testUtils) throws IOException {
  // Here we create all different type of replicas and add it
  // to volume map. 
  // Created all type of ReplicaInfo, each under Blkpool corresponding volume
  long id = 1; // This variable is used as both blockId and genStamp
  for (String bpId: bpList) {
    for (FsVolumeSpi volume: volumes) {
      ExtendedBlock eb = new ExtendedBlock(bpId, id, 1, id);
      testUtils.createFinalizedReplica(volume, eb);
      id++;

      eb = new ExtendedBlock(bpId, id, 1, id);
      testUtils.createRBW(volume, eb);
      id++;

      eb = new ExtendedBlock(bpId, id, 1, id);
      testUtils.createReplicaWaitingToBeRecovered(volume, eb);
      id++;

      eb = new ExtendedBlock(bpId, id, 1, id);
      testUtils.createReplicaInPipeline(volume, eb);
      id++;
    }
  }
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
 * Check that the permissions of the local DN directories are as expected.
 */
@Test
public void testLocalDirs() throws Exception {
  Configuration conf = new Configuration();
  final String permStr = conf.get(
    DFSConfigKeys.DFS_DATANODE_DATA_DIR_PERMISSION_KEY);
  FsPermission expected = new FsPermission(permStr);

  // Check permissions on directories in 'dfs.datanode.data.dir'
  FileSystem localFS = FileSystem.getLocal(conf);
  for (DataNode dn : cluster.getDataNodes()) {
    try (FsDatasetSpi.FsVolumeReferences volumes =
        dn.getFSDataset().getFsVolumeReferences()) {
      for (FsVolumeSpi vol : volumes) {
        String dir = vol.getBasePath();
        Path dataDir = new Path(dir);
        FsPermission actual = localFS.getFileStatus(dataDir).getPermission();
        assertEquals("Permission for dir: " + dataDir + ", is " + actual +
            ", while expected is " + expected, expected, actual);
      }
    }
  }
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
long getRemaining() throws IOException {
  long remaining = 0L;
  for (FsVolumeSpi vol : volumes.get()) {
    try (FsVolumeReference ref = vol.obtainReference()) {
      remaining += vol.getAvailable();
    } catch (ClosedChannelException e) {
      // ignore
    }
  }
  return remaining;
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/** Block is not found on the disk */
private void addDifference(LinkedList<ScanInfo> diffRecord,
                           Stats statsRecord, long blockId,
                           FsVolumeSpi vol) {
  statsRecord.missingBlockFile++;
  statsRecord.missingMetaFile++;
  diffRecord.add(new ScanInfo(blockId, null, null, vol));
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/** Is the given volume still valid in the dataset? */
private static boolean isValid(final FsDatasetSpi<?> dataset,
    final FsVolumeSpi volume) {
  for (FsVolumeSpi vol : dataset.getVolumes()) {
    if (vol == volume) {
      return true;
    }
  }
  return false;
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
 * Report a bad block which is hosted on the local DN.
 */
public void reportBadBlocks(ExtendedBlock block) throws IOException{
  BPOfferService bpos = getBPOSForBlock(block);
  FsVolumeSpi volume = getFSDataset().getVolume(block);
  bpos.reportBadBlocks(
      block, volume.getStorageID(), volume.getStorageType());
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
private void reportBadBlock(final BPOfferService bpos,
    final ExtendedBlock block, final String msg) {
  FsVolumeSpi volume = getFSDataset().getVolume(block);
  bpos.reportBadBlocks(
      block, volume.getStorageID(), volume.getStorageType());
  LOG.warn(msg);
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
* Set up a scanner for the given block pool and volume.
*
* @param ref              A reference to the volume.
*/
public synchronized void addVolumeScanner(FsVolumeReference ref) {
  boolean success = false;
  try {
    FsVolumeSpi volume = ref.getVolume();
    if (!isEnabled()) {
      LOG.debug("Not adding volume scanner for {}, because the block " +
          "scanner is disabled.", volume.getBasePath());
      return;
    }
    VolumeScanner scanner = scanners.get(volume.getStorageID());
    if (scanner != null) {
      LOG.error("Already have a scanner for volume {}.",
          volume.getBasePath());
      return;
    }
    LOG.debug("Adding scanner for volume {} (StorageID {})",
        volume.getBasePath(), volume.getStorageID());
    scanner = new VolumeScanner(conf, datanode, ref);
    scanner.start();
    scanners.put(volume.getStorageID(), scanner);
    success = true;
  } finally {
    if (!success) {
      // If we didn't create a new VolumeScanner object, we don't
      // need this reference to the volume.
      IOUtils.cleanup(null, ref);
    }
  }
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
public void handle(ExtendedBlock block, IOException e) {
  FsVolumeSpi volume = scanner.volume;
  if (e == null) {
    LOG.trace("Successfully scanned {} on {}", block, volume.getBasePath());
    return;
  }
  // If the block does not exist anymore, then it's not an error.
  if (!volume.getDataset().contains(block)) {
    LOG.debug("Volume {}: block {} is no longer in the dataset.",
        volume.getBasePath(), block);
    return;
  }
  // If the block exists, the exception may due to a race with write:
  // The BlockSender got an old block path in rbw. BlockReceiver removed
  // the rbw block from rbw to finalized but BlockSender tried to open the
  // file before BlockReceiver updated the VolumeMap. The state of the
  // block can be changed again now, so ignore this error here. If there
  // is a block really deleted by mistake, DirectoryScan should catch it.
  if (e instanceof FileNotFoundException ) {
    LOG.info("Volume {}: verification failed for {} because of " +
            "FileNotFoundException.  This may be due to a race with write.",
        volume.getBasePath(), block);
    return;
  }
  LOG.warn("Reporting bad {} on {}", block, volume.getBasePath());
  try {
    scanner.datanode.reportBadBlocks(block);
  } catch (IOException ie) {
    // This is bad, but not bad enough to shut down the scanner.
    LOG.warn("Cannot report bad " + block.getBlockId(), e);
  }
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
private void setVolumeFull(DataNode dn, StorageType type) {
  List<? extends FsVolumeSpi> volumes = dn.getFSDataset().getVolumes();
  for (FsVolumeSpi v : volumes) {
    FsVolumeImpl volume = (FsVolumeImpl) v;
    if (volume.getStorageType() == type) {
      LOG.info("setCapacity to 0 for [" + volume.getStorageType() + "]"
          + volume.getStorageID());
      volume.setCapacityForTesting(0);
    }
  }
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
 * Make sure at least one non-transient volume has a saved copy of the replica.
 * An infinite loop is used to ensure the async lazy persist tasks are completely
 * done before verification. Caller of ensureLazyPersistBlocksAreSaved expects
 * either a successful pass or timeout failure.
 */
protected final void ensureLazyPersistBlocksAreSaved(
    LocatedBlocks locatedBlocks) throws IOException, InterruptedException {
  final String bpid = cluster.getNamesystem().getBlockPoolId();
  List<? extends FsVolumeSpi> volumes =
    cluster.getDataNodes().get(0).getFSDataset().getVolumes();
  final Set<Long> persistedBlockIds = new HashSet<Long>();

  while (persistedBlockIds.size() < locatedBlocks.getLocatedBlocks().size()) {
    // Take 1 second sleep before each verification iteration
    Thread.sleep(1000);

    for (LocatedBlock lb : locatedBlocks.getLocatedBlocks()) {
      for (FsVolumeSpi v : volumes) {
        if (v.isTransientStorage()) {
          continue;
        }

        FsVolumeImpl volume = (FsVolumeImpl) v;
        File lazyPersistDir = volume.getBlockPoolSlice(bpid).getLazypersistDir();

        long blockId = lb.getBlock().getBlockId();
        File targetDir =
          DatanodeUtil.idToBlockDir(lazyPersistDir, blockId);
        File blockFile = new File(targetDir, lb.getBlock().getBlockName());
        if (blockFile.exists()) {
          // Found a persisted copy for this block and added to the Set
          persistedBlockIds.add(blockId);
        }
      }
    }
  }

  // We should have found a persisted copy for each located block.
  assertThat(persistedBlockIds.size(), is(locatedBlocks.getLocatedBlocks().size()));
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
protected final boolean verifyDeletedBlocks(LocatedBlocks locatedBlocks)
    throws IOException, InterruptedException {

  LOG.info("Verifying replica has no saved copy after deletion.");
  triggerBlockReport();

  while(
    DataNodeTestUtils.getPendingAsyncDeletions(cluster.getDataNodes().get(0))
      > 0L){
    Thread.sleep(1000);
  }

  final String bpid = cluster.getNamesystem().getBlockPoolId();
  List<? extends FsVolumeSpi> volumes =
    cluster.getDataNodes().get(0).getFSDataset().getVolumes();

  // Make sure deleted replica does not have a copy on either finalized dir of
  // transient volume or finalized dir of non-transient volume
  for (FsVolumeSpi v : volumes) {
    FsVolumeImpl volume = (FsVolumeImpl) v;
    File targetDir = (v.isTransientStorage()) ?
        volume.getBlockPoolSlice(bpid).getFinalizedDir() :
        volume.getBlockPoolSlice(bpid).getLazypersistDir();
    if (verifyBlockDeletedFromDir(targetDir, locatedBlocks) == false) {
      return false;
    }
  }
  return true;
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
 *
 * @param blockSize
 * @param perVolumeCapacity limit the capacity of each volume to the given
 *                          value. If negative, then don't limit.
 * @throws IOException
 */
private void startCluster(int blockSize, int numDatanodes, long perVolumeCapacity) throws IOException {
  initConfig(blockSize);

  cluster = new MiniDFSCluster
      .Builder(conf)
      .storagesPerDatanode(STORAGES_PER_DATANODE)
      .numDataNodes(numDatanodes)
      .build();
  fs = cluster.getFileSystem();
  client = fs.getClient();
  cluster.waitActive();

  if (perVolumeCapacity >= 0) {
    for (DataNode dn : cluster.getDataNodes()) {
      for (FsVolumeSpi volume : dn.getFSDataset().getVolumes()) {
        ((FsVolumeImpl) volume).setCapacityForTesting(perVolumeCapacity);
      }
    }
  }

  if (numDatanodes == 1) {
    List<? extends FsVolumeSpi> volumes =
        cluster.getDataNodes().get(0).getFSDataset().getVolumes();
    assertThat(volumes.size(), is(1));
    singletonVolume = ((FsVolumeImpl) volumes.get(0));
  }
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/** Get the FsVolume on the given basePath */
private FsVolumeImpl getVolume(DataNode dn, File basePath) {
  for (FsVolumeSpi vol : dn.getFSDataset().getVolumes()) {
    if (vol.getBasePath().equals(basePath.getPath())) {
      return (FsVolumeImpl)vol;
    }
  }
  return null;
}
 

import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeSpi; //导入依赖的package包/类
/**
 * Duplicate the given block on all volumes.
 * @param blockId
 * @throws IOException
 */
private void duplicateBlock(long blockId) throws IOException {
  synchronized (fds) {
    ReplicaInfo b = FsDatasetTestUtil.fetchReplicaInfo(fds, bpid, blockId);
    for (FsVolumeSpi v : fds.getVolumes()) {
      if (v.getStorageID().equals(b.getVolume().getStorageID())) {
        continue;
      }

      // Volume without a copy of the block. Make a copy now.
      File sourceBlock = b.getBlockFile();
      File sourceMeta = b.getMetaFile();
      String sourceRoot = b.getVolume().getBasePath();
      String destRoot = v.getBasePath();

      String relativeBlockPath = new File(sourceRoot).toURI().relativize(sourceBlock.toURI()).getPath();
      String relativeMetaPath = new File(sourceRoot).toURI().relativize(sourceMeta.toURI()).getPath();

      File destBlock = new File(destRoot, relativeBlockPath);
      File destMeta = new File(destRoot, relativeMetaPath);

      destBlock.getParentFile().mkdirs();
      FileUtils.copyFile(sourceBlock, destBlock);
      FileUtils.copyFile(sourceMeta, destMeta);

      if (destBlock.exists() && destMeta.exists()) {
        LOG.info("Copied " + sourceBlock + " ==> " + destBlock);
        LOG.info("Copied " + sourceMeta + " ==> " + destMeta);
      }
    }
  }
}