@Override
public void close() throws Exception {
  if (readaheadRequest != null) {
    readaheadRequest.cancel();
  }
  if (manageOsCache && getEndOffset() - getStartOffset() > 0) {
    try {
      NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(identifier,
          fd,
          getStartOffset(), getEndOffset() - getStartOffset(),
          NativeIO.POSIX.POSIX_FADV_DONTNEED);
    } catch (Throwable t) {
      LOG.warn("Failed to manage OS cache for " + identifier, t);
    }
  }
  super.close();
}
 

@Override
public void run() {
  if (canceled) return;
  // There's a very narrow race here that the file will close right at
  // this instant. But if that happens, we'll likely receive an EBADF
  // error below, and see that it's canceled, ignoring the error.
  // It's also possible that we'll end up requesting readahead on some
  // other FD, which may be wasted work, but won't cause a problem.
  try {
    NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(identifier,
        fd, off, len, NativeIO.POSIX.POSIX_FADV_WILLNEED);
  } catch (IOException ioe) {
    if (canceled) {
      // no big deal - the reader canceled the request and closed
      // the file.
      return;
    }
    LOG.warn("Failed readahead on " + identifier,
        ioe);
  }
}
 

public void run() {
  if (canceled) return;
  // There's a very narrow race here that the file will close right at
  // this instant. But if that happens, we'll likely receive an EBADF
  // error below, and see that it's canceled, ignoring the error.
  // It's also possible that we'll end up requesting readahead on some
  // other FD, which may be wasted work, but won't cause a problem.
  try {
    NativeIO.posixFadviseIfPossible(fd, off, len,
        NativeIO.POSIX_FADV_WILLNEED);
  } catch (IOException ioe) {
    if (canceled) {
      // no big deal - the reader canceled the request and closed
      // the file.
      return;
    }
    LOG.warn("Failed readahead on " + identifier,
        ioe);
  }
}
 

/**
 * Same as openFSDataInputStream except that it will run even if security is
 * off. This is used by unit tests.
 */
@VisibleForTesting
protected static FSDataInputStream forceSecureOpenFSDataInputStream(
    File file,
    String expectedOwner, String expectedGroup) throws IOException {
  final FSDataInputStream in =
      rawFilesystem.open(new Path(file.getAbsolutePath()));
  boolean success = false;
  try {
    Stat stat = NativeIO.POSIX.getFstat(in.getFileDescriptor());
    checkStat(file, stat.getOwner(), stat.getGroup(), expectedOwner,
        expectedGroup);
    success = true;
    return in;
  } finally {
    if (!success) {
      in.close();
    }
  }
}
 

/**
 * Bump a replica's generation stamp to a new one.
 * Its on-disk meta file name is renamed to be the new one too.
 * 
 * @param replicaInfo a replica
 * @param newGS new generation stamp
 * @throws IOException if rename fails
 */
private void bumpReplicaGS(ReplicaInfo replicaInfo, 
    long newGS) throws IOException { 
  long oldGS = replicaInfo.getGenerationStamp();
  File oldmeta = replicaInfo.getMetaFile();
  replicaInfo.setGenerationStamp(newGS);
  File newmeta = replicaInfo.getMetaFile();

  // rename meta file to new GS
  if (LOG.isDebugEnabled()) {
    LOG.debug("Renaming " + oldmeta + " to " + newmeta);
  }
  try {
    NativeIO.renameTo(oldmeta, newmeta);
  } catch (IOException e) {
    replicaInfo.setGenerationStamp(oldGS); // restore old GS
    throw new IOException("Block " + replicaInfo + " reopen failed. " +
                          " Unable to move meta file  " + oldmeta +
                          " to " + newmeta, e);
  }
}
 

public static HdfsConfiguration initZeroCopyTest() {
  Assume.assumeTrue(NativeIO.isAvailable());
  Assume.assumeTrue(SystemUtils.IS_OS_UNIX);
  HdfsConfiguration conf = new HdfsConfiguration();
  conf.setBoolean(DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_KEY, true);
  conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, BLOCK_SIZE);
  conf.setInt(DFSConfigKeys.DFS_CLIENT_MMAP_CACHE_SIZE, 3);
  conf.setLong(DFSConfigKeys.DFS_CLIENT_MMAP_CACHE_TIMEOUT_MS, 100);
  conf.set(DFSConfigKeys.DFS_DOMAIN_SOCKET_PATH_KEY,
      new File(sockDir.getDir(),
        "TestRequestMmapAccess._PORT.sock").getAbsolutePath());
  conf.setBoolean(DFSConfigKeys.
      DFS_CLIENT_READ_SHORTCIRCUIT_SKIP_CHECKSUM_KEY, true);
  conf.setLong(DFS_HEARTBEAT_INTERVAL_KEY, 1);
  conf.setLong(DFS_CACHEREPORT_INTERVAL_MSEC_KEY, 1000);
  conf.setLong(DFS_NAMENODE_PATH_BASED_CACHE_REFRESH_INTERVAL_MS, 1000);
  return conf;
}
 

@Override
public void close() throws Exception {
  if (readaheadRequest != null) {
    readaheadRequest.cancel();
  }
  if (PullServerUtil.isNativeIOPossible() && manageOsCache && endOffset() - startOffset() > 0) {
    try {
      PullServerUtil.posixFadviseIfPossible(identifier,
          fd,
          startOffset(), endOffset() - startOffset(),
          NativeIO.POSIX.POSIX_FADV_DONTNEED);
    } catch (Throwable t) {
      LOG.warn("Failed to manage OS cache for " + identifier, t);
    }
  }
  super.close();
}
 

protected boolean mkOneDirWithMode(Path p, File p2f, FsPermission permission)
    throws IOException {
  if (permission == null) {
    return p2f.mkdir();
  } else {
    if (Shell.WINDOWS && NativeIO.isAvailable()) {
      try {
        NativeIO.Windows.createDirectoryWithMode(p2f, permission.toShort());
        return true;
      } catch (IOException e) {
        if (LOG.isDebugEnabled()) {
          LOG.debug(String.format(
              "NativeIO.createDirectoryWithMode error, path = %s, mode = %o",
              p2f, permission.toShort()), e);
        }
        return false;
      }
    } else {
      boolean b = p2f.mkdir();
      if (b) {
        setPermission(p, permission);
      }
      return b;
    }
  }
}
 

/**
 * Run testCacheAndUncacheBlock with some failures injected into the mlock
 * call.  This tests the ability of the NameNode to resend commands.
 */
@Test(timeout=600000)
public void testCacheAndUncacheBlockWithRetries() throws Exception {
  // We don't have to save the previous cacheManipulator
  // because it will be reinstalled by the @After function.
  NativeIO.POSIX.setCacheManipulator(new NoMlockCacheManipulator() {
    private final Set<String> seenIdentifiers = new HashSet<String>();
    
    @Override
    public void mlock(String identifier,
        ByteBuffer mmap, long length) throws IOException {
      if (seenIdentifiers.contains(identifier)) {
        // mlock succeeds the second time.
        LOG.info("mlocking " + identifier);
        return;
      }
      seenIdentifiers.add(identifier);
      throw new IOException("injecting IOException during mlock of " +
          identifier);
    }
  });
  testCacheAndUncacheBlock();
}
 

private LocalFSFileOutputStream(Path f, boolean append,
    FsPermission permission) throws IOException {
  File file = pathToFile(f);
  if (permission == null) {
    this.fos = new FileOutputStream(file, append);
  } else {
    if (Shell.WINDOWS && NativeIO.isAvailable()) {
      this.fos = NativeIO.Windows.createFileOutputStreamWithMode(file,
          append, permission.toShort());
    } else {
      this.fos = new FileOutputStream(file, append);
      boolean success = false;
      try {
        setPermission(f, permission);
        success = true;
      } finally {
        if (!success) {
          IOUtils.cleanup(LOG, this.fos);
        }
      }
    }
  }
}
 

@Override
public void run() {
  if (canceled) return;
  // There's a very narrow race here that the file will close right at
  // this instant. But if that happens, we'll likely receive an EBADF
  // error below, and see that it's canceled, ignoring the error.
  // It's also possible that we'll end up requesting readahead on some
  // other FD, which may be wasted work, but won't cause a problem.
  try {
    NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(identifier,
        fd, off, len, NativeIO.POSIX.POSIX_FADV_WILLNEED);
  } catch (IOException ioe) {
    if (canceled) {
      // no big deal - the reader canceled the request and closed
      // the file.
      return;
    }
    LOG.warn("Failed readahead on " + identifier,
        ioe);
  }
}
 

/**
 * Same as openFSDataInputStream except that it will run even if security is
 * off. This is used by unit tests.
 */
@VisibleForTesting
protected static FSDataInputStream forceSecureOpenFSDataInputStream(
    File file,
    String expectedOwner, String expectedGroup) throws IOException {
  final FSDataInputStream in =
      rawFilesystem.open(new Path(file.getAbsolutePath()));
  boolean success = false;
  try {
    Stat stat = NativeIO.POSIX.getFstat(in.getFileDescriptor());
    checkStat(file, stat.getOwner(), stat.getGroup(), expectedOwner,
        expectedGroup);
    success = true;
    return in;
  } finally {
    if (!success) {
      in.close();
    }
  }
}
 

/**
 * Same as openForRead() except that it will run even if security is off.
 * This is used by unit tests.
 */
@VisibleForTesting
protected static FileInputStream forceSecureOpenForRead(File f, String expectedOwner,
    String expectedGroup) throws IOException {

  FileInputStream fis = new FileInputStream(f);
  boolean success = false;
  try {
    Stat stat = NativeIO.POSIX.getFstat(fis.getFD());
    checkStat(f, stat.getOwner(), stat.getGroup(), expectedOwner,
        expectedGroup);
    success = true;
    return fis;
  } finally {
    if (!success) {
      fis.close();
    }
  }
}
 

@Override
public void close() throws Exception {
  if (readaheadRequest != null) {
    readaheadRequest.cancel();
  }
  if (manageOsCache && getEndOffset() - getStartOffset() > 0) {
    try {
      NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(identifier,
          fd,
          getStartOffset(), getEndOffset() - getStartOffset(),
          NativeIO.POSIX.POSIX_FADV_DONTNEED);
    } catch (Throwable t) {
      LOG.warn("Failed to manage OS cache for " + identifier, t);
    }
  }
  super.close();
}
 

/**
 * Run testCacheAndUncacheBlock with some failures injected into the mlock
 * call.  This tests the ability of the NameNode to resend commands.
 */
@Test(timeout=600000)
public void testCacheAndUncacheBlockWithRetries() throws Exception {
  // We don't have to save the previous cacheManipulator
  // because it will be reinstalled by the @After function.
  NativeIO.POSIX.setCacheManipulator(new NoMlockCacheManipulator() {
    private final Set<String> seenIdentifiers = new HashSet<String>();
    
    @Override
    public void mlock(String identifier,
        ByteBuffer mmap, long length) throws IOException {
      if (seenIdentifiers.contains(identifier)) {
        // mlock succeeds the second time.
        LOG.info("mlocking " + identifier);
        return;
      }
      seenIdentifiers.add(identifier);
      throw new IOException("injecting IOException during mlock of " +
          identifier);
    }
  });
  testCacheAndUncacheBlock();
}
 

/**
 * Same as openForRandomRead except that it will run even if security is off.
 * This is used by unit tests.
 */
@VisibleForTesting
protected static RandomAccessFile forceSecureOpenForRandomRead(File f,
    String mode, String expectedOwner, String expectedGroup)
    throws IOException {
  RandomAccessFile raf = new RandomAccessFile(f, mode);
  boolean success = false;
  try {
    Stat stat = NativeIO.POSIX.getFstat(raf.getFD());
    checkStat(f, stat.getOwner(), stat.getGroup(), expectedOwner,
        expectedGroup);
    success = true;
    return raf;
  } finally {
    if (!success) {
      raf.close();
    }
  }
}
 

/**
 * Bump a replica's generation stamp to a new one.
 * Its on-disk meta file name is renamed to be the new one too.
 * 
 * @param replicaInfo a replica
 * @param newGS new generation stamp
 * @throws IOException if rename fails
 */
private void bumpReplicaGS(ReplicaInfo replicaInfo, 
    long newGS) throws IOException { 
  long oldGS = replicaInfo.getGenerationStamp();
  File oldmeta = replicaInfo.getMetaFile();
  replicaInfo.setGenerationStamp(newGS);
  File newmeta = replicaInfo.getMetaFile();

  // rename meta file to new GS
  if (LOG.isDebugEnabled()) {
    LOG.debug("Renaming " + oldmeta + " to " + newmeta);
  }
  try {
    NativeIO.renameTo(oldmeta, newmeta);
  } catch (IOException e) {
    replicaInfo.setGenerationStamp(oldGS); // restore old GS
    throw new IOException("Block " + replicaInfo + " reopen failed. " +
                          " Unable to move meta file  " + oldmeta +
                          " to " + newmeta, e);
  }
}
 

/**
 * Load the block.
 *
 * mmap and mlock the block, and then verify its checksum.
 *
 * @param length         The current length of the block.
 * @param blockIn        The block input stream.  Should be positioned at the
 *                       start.  The caller must close this.
 * @param metaIn         The meta file input stream.  Should be positioned at
 *                       the start.  The caller must close this.
 * @param blockFileName  The block file name, for logging purposes.
 *
 * @return               The Mappable block.
 */
public static MappableBlock load(long length,
    FileInputStream blockIn, FileInputStream metaIn,
    String blockFileName) throws IOException {
  MappableBlock mappableBlock = null;
  MappedByteBuffer mmap = null;
  FileChannel blockChannel = null;
  try {
    blockChannel = blockIn.getChannel();
    if (blockChannel == null) {
      throw new IOException("Block InputStream has no FileChannel.");
    }
    mmap = blockChannel.map(MapMode.READ_ONLY, 0, length);
    NativeIO.POSIX.getCacheManipulator().mlock(blockFileName, mmap, length);
    verifyChecksum(length, metaIn, blockChannel, blockFileName);
    mappableBlock = new MappableBlock(mmap, length);
  } finally {
    IOUtils.closeQuietly(blockChannel);
    if (mappableBlock == null) {
      if (mmap != null) {
        NativeIO.POSIX.munmap(mmap); // unmapping also unlocks
      }
    }
  }
  return mappableBlock;
}
 

private LocalFSFileOutputStream(Path f, boolean append,
    FsPermission permission) throws IOException {
  File file = pathToFile(f);
  if (permission == null) {
    this.fos = new FileOutputStream(file, append);
  } else {
    if (Shell.WINDOWS && NativeIO.isAvailable()) {
      this.fos = NativeIO.Windows.createFileOutputStreamWithMode(file,
          append, permission.toShort());
    } else {
      this.fos = new FileOutputStream(file, append);
      boolean success = false;
      try {
        setPermission(f, permission);
        success = true;
      } finally {
        if (!success) {
          IOUtils.cleanup(LOG, this.fos);
        }
      }
    }
  }
}
 

public static HdfsConfiguration initZeroCopyTest() {
  Assume.assumeTrue(NativeIO.isAvailable());
  Assume.assumeTrue(SystemUtils.IS_OS_UNIX);
  HdfsConfiguration conf = new HdfsConfiguration();
  conf.setBoolean(DFSConfigKeys.DFS_CLIENT_READ_SHORTCIRCUIT_KEY, true);
  conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, BLOCK_SIZE);
  conf.setInt(DFSConfigKeys.DFS_CLIENT_MMAP_CACHE_SIZE, 3);
  conf.setLong(DFSConfigKeys.DFS_CLIENT_MMAP_CACHE_TIMEOUT_MS, 100);
  conf.set(DFSConfigKeys.DFS_DOMAIN_SOCKET_PATH_KEY,
      new File(sockDir.getDir(),
        "TestRequestMmapAccess._PORT.sock").getAbsolutePath());
  conf.setBoolean(DFSConfigKeys.
      DFS_CLIENT_READ_SHORTCIRCUIT_SKIP_CHECKSUM_KEY, true);
  conf.setLong(DFS_HEARTBEAT_INTERVAL_KEY, 1);
  conf.setLong(DFS_CACHEREPORT_INTERVAL_MSEC_KEY, 1000);
  conf.setLong(DFS_NAMENODE_PATH_BASED_CACHE_REFRESH_INTERVAL_MS, 1000);
  return conf;
}
 

/**
 * Writes to .container file.
 *
 * @param containerFile container file name
 * @param isCreate True if creating a new file. False is updating an
 *                 existing container file.
 * @throws StorageContainerException
 */
private void writeToContainerFile(File containerFile, boolean isCreate)
    throws StorageContainerException {
  File tempContainerFile = null;
  long containerId = containerData.getContainerID();
  try {
    tempContainerFile = createTempFile(containerFile);
    ContainerDataYaml.createContainerFile(
        ContainerType.KeyValueContainer, containerData, tempContainerFile);

    // NativeIO.renameTo is an atomic function. But it might fail if the
    // container file already exists. Hence, we handle the two cases
    // separately.
    if (isCreate) {
      NativeIO.renameTo(tempContainerFile, containerFile);
    } else {
      Files.move(tempContainerFile.toPath(), containerFile.toPath(),
          StandardCopyOption.REPLACE_EXISTING);
    }

  } catch (IOException ex) {
    throw new StorageContainerException("Error while creating/ updating " +
        ".container file. ContainerID: " + containerId, ex,
        CONTAINER_FILES_CREATE_ERROR);
  } finally {
    if (tempContainerFile != null && tempContainerFile.exists()) {
      if (!tempContainerFile.delete()) {
        LOG.warn("Unable to delete container temporary file: {}.",
            tempContainerFile.getAbsolutePath());
      }
    }
  }
}
 

/**
 * Platform independent implementation for {@link File#canWrite()}
 * @param f input file
 * @return On Unix, same as {@link File#canWrite()}
 *         On Windows, true if process has write access on the path
 */
public static boolean canWrite(File f) {
  if (Shell.WINDOWS) {
    try {
      return NativeIO.Windows.access(f.getCanonicalPath(),
          NativeIO.Windows.AccessRight.ACCESS_WRITE);
    } catch (IOException e) {
      return false;
    }
  } else {
    return f.canWrite();
  }
}
 

/**
 * Load file permission information (UNIX symbol rwxrwxrwx, sticky bit info).
 *
 * To improve peformance, give priority to native stat() call. First try get
 * permission information by using native JNI call then fall back to use non
 * native (ProcessBuilder) call in case native lib is not loaded or native
 * call is not successful
 */
private synchronized void loadPermissionInfo() {
  if (!isPermissionLoaded() && NativeIO.isAvailable()) {
    try {
      loadPermissionInfoByNativeIO();
    } catch (IOException ex) {
      LOG.debug("Native call failed", ex);
    }
  }

  if (!isPermissionLoaded()) {
    loadPermissionInfoByNonNativeIO();
  }
}
 

public void submitSyncFileRangeRequest(FsVolumeImpl volume,
    final FileDescriptor fd, final long offset, final long nbytes,
    final int flags) {
  execute(volume.getCurrentDir(), new Runnable() {
    @Override
    public void run() {
      try {
        NativeIO.POSIX.syncFileRangeIfPossible(fd, offset, nbytes, flags);
      } catch (NativeIOException e) {
        LOG.warn("sync_file_range error", e);
      }
    }
  });
}
 

@Override // FsDatasetSpi
public InputStream getBlockInputStream(ExtendedBlock b,
    long seekOffset) throws IOException {
  File blockFile = getBlockFileNoExistsCheck(b, true);
  if (isNativeIOAvailable) {
    return NativeIO.getShareDeleteFileInputStream(blockFile, seekOffset);
  } else {
    try {
      return openAndSeek(blockFile, seekOffset);
    } catch (FileNotFoundException fnfe) {
      throw new IOException("Block " + b + " is not valid. " +
          "Expected block file at " + blockFile + " does not exist.");
    }
  }
}
 

@Override
public void close() {
  if (mmap != null) {
    NativeIO.POSIX.munmap(mmap);
    mmap = null;
  }
}
 

/**
 * Manage the OS buffer cache by performing read-ahead
 * and drop-behind.
 */
private void manageOsCache() throws IOException {
  // We can't manage the cache for this block if we don't have a file
  // descriptor to work with.
  if (blockInFd == null) return;

  // Perform readahead if necessary
  if ((readaheadLength > 0) && (datanode.readaheadPool != null) &&
        (alwaysReadahead || isLongRead())) {
    curReadahead = datanode.readaheadPool.readaheadStream(
        clientTraceFmt, blockInFd, offset, readaheadLength, Long.MAX_VALUE,
        curReadahead);
  }

  // Drop what we've just read from cache, since we aren't
  // likely to need it again
  if (dropCacheBehindAllReads ||
      (dropCacheBehindLargeReads && isLongRead())) {
    long nextCacheDropOffset = lastCacheDropOffset + CACHE_DROP_INTERVAL_BYTES;
    if (offset >= nextCacheDropOffset) {
      long dropLength = offset - lastCacheDropOffset;
      NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(
          block.getBlockName(), blockInFd, lastCacheDropOffset,
          dropLength, NativeIO.POSIX.POSIX_FADV_DONTNEED);
      lastCacheDropOffset = offset;
    }
  }
}
 

public static void rename(File from, File to) throws IOException {
  try {
    NativeIO.renameTo(from, to);
  } catch (NativeIOException e) {
    throw new IOException("Failed to rename " + from.getCanonicalPath()
      + " to " + to.getCanonicalPath() + " due to failure in native rename. "
      + e.toString());
  }
}
 

private static void execSetPermission(File f,
                                      FsPermission permission
)  throws IOException {
  if (NativeIO.isAvailable()) {
    NativeIO.POSIX.chmod(f.getCanonicalPath(), permission.toShort());
  } else {
    execCommand(f, Shell.getSetPermissionCommand(
        String.format("%04o", permission.toShort()), false));
  }
}
 

@Override
public void close() throws IOException {
  boolean triedToClose = false, success = false;
  try {
    flush();
    ((FileOutputStream)out).getChannel().force(true);

    triedToClose = true;
    super.close();
    success = true;
  } finally {
    if (success) {
      boolean renamed = tmpFile.renameTo(origFile);
      if (!renamed) {
        // On windows, renameTo does not replace.
        if (origFile.exists() && !origFile.delete()) {
          throw new IOException("Could not delete original file " + origFile);
        }
        try {
          NativeIO.renameTo(tmpFile, origFile);
        } catch (NativeIOException e) {
          throw new IOException("Could not rename temporary file " + tmpFile
            + " to " + origFile + " due to failure in native rename. "
            + e.toString());
        }
      }
    } else {
      if (!triedToClose) {
        // If we failed when flushing, try to close it to not leak an FD
        IOUtils.closeStream(out);
      }
      // close wasn't successful, try to delete the tmp file
      if (!tmpFile.delete()) {
        LOG.warn("Unable to delete tmp file " + tmpFile);
      }
    }
  }
}