Skip to content

HADOOP-17028. ViewFS should initialize mounted target filesystems lazily #3218

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: branch-2.10
Choose a base branch
from
Open

HADOOP-17028. ViewFS should initialize mounted target filesystems lazily #3218

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 41 additions & 10 deletions hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
*/
package org.apache.hadoop.fs.viewfs;

import com.google.common.base.Function;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;
Expand Down Expand Up @@ -158,8 +159,10 @@ void addLink(final String pathComponent, final INodeLink<T> link)
static class INodeLink<T> extends INode<T> {
final boolean isMergeLink; // true if MergeLink
final URI[] targetDirLinkList;
final T targetFileSystem; // file system object created from the link.

private T targetFileSystem; // file system object created from the link.
// Function to initialize file system. Only applicable for simple links
private Function<URI, T> fileSystemInitMethod;
private final Object lock = new Object();
/**
* Construct a mergeLink
*/
Expand All @@ -175,12 +178,14 @@ static class INodeLink<T> extends INode<T> {
* Construct a simple link (i.e. not a mergeLink)
*/
INodeLink(final String pathToNode, final UserGroupInformation aUgi,
final T targetFs, final URI aTargetDirLink) {
Function<URI, T> createFileSystemMethod,
final URI aTargetDirLink) {
super(pathToNode, aUgi);
targetFileSystem = targetFs;
targetFileSystem = null;
targetDirLinkList = new URI[1];
targetDirLinkList[0] = aTargetDirLink;
isMergeLink = false;
this.fileSystemInitMethod = createFileSystemMethod;
}

/**
Expand All @@ -196,6 +201,33 @@ Path getTargetLink() {
}
return new Path(result.toString());
}

/**
* Get the instance of FileSystem to use, creating one if needed.
* @return An Initialized instance of T
* @throws IOException
*/
public T getTargetFileSystem() throws IOException {
if (targetFileSystem != null) {
return targetFileSystem;
}
// For non NFLY and MERGE links, we initialize the FileSystem when the
// corresponding mount path is accessed.
if (targetDirLinkList.length == 1) {
synchronized (lock) {
if (targetFileSystem != null) {
return targetFileSystem;
}
targetFileSystem = fileSystemInitMethod.apply(targetDirLinkList[0]);
if (targetFileSystem == null) {
throw new IOException(
"Could not initialize target File System for URI : " +
targetDirLinkList[0]);
}
}
}
return targetFileSystem;
}
}


Expand Down Expand Up @@ -258,7 +290,7 @@ private void createLink(final String src, final String target,
getTargetFileSystem(targetsListURI), targetsListURI);
} else {
newLink = new INodeLink<T>(fullPath, aUgi,
getTargetFileSystem(new URI(target)), new URI(target));
initAndGetTargetFs(), new URI(target));
}
curInode.addLink(iPath, newLink);
mountPoints.add(new MountPoint<T>(src, newLink));
Expand All @@ -267,14 +299,13 @@ private void createLink(final String src, final String target,
/**
* Below the "public" methods of InodeTree
*/

/**
* The user of this class must subclass and implement the following
* 3 abstract methods.
* @throws IOException
*/
protected abstract T getTargetFileSystem(final URI uri)
throws UnsupportedFileSystemException, URISyntaxException, IOException;
protected abstract Function<URI, T> initAndGetTargetFs();

protected abstract T getTargetFileSystem(final INodeDir<T> dir)
throws URISyntaxException;
Expand Down Expand Up @@ -385,7 +416,7 @@ boolean isInternalDir() {
* @throws FileNotFoundException
*/
ResolveResult<T> resolve(final String p, final boolean resolveLastComponent)
throws FileNotFoundException {
throws IOException {
// TO DO: - more efficient to not split the path, but simply compare
String[] path = breakIntoPathComponents(p);
if (path.length <= 1) { // special case for when path is "/"
Expand Down Expand Up @@ -422,7 +453,7 @@ ResolveResult<T> resolve(final String p, final boolean resolveLastComponent)
}
final ResolveResult<T> res =
new ResolveResult<T>(ResultKind.isExternalDir,
link.targetFileSystem, nextInode.fullPath, remainingPath);
link.getTargetFileSystem(), nextInode.fullPath, remainingPath);
return res;
} else if (nextInode instanceof INodeDir) {
curInode = (INodeDir<T>) nextInode;
Expand Down
85 changes: 65 additions & 20 deletions ...ommon-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,12 @@
import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE;
import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE_DEFAULT;

import com.google.common.base.Function;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.security.PrivilegedExceptionAction;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
Expand Down Expand Up @@ -237,24 +239,40 @@ public void initialize(final URI theUri, final Configuration conf)
config = conf;
enableInnerCache = config.getBoolean(CONFIG_VIEWFS_ENABLE_INNER_CACHE,
CONFIG_VIEWFS_ENABLE_INNER_CACHE_DEFAULT);
final InnerCache innerCache = new InnerCache();
cache = new InnerCache();
// Now build client side view (i.e. client side mount table) from config.
final String authority = theUri.getAuthority();
try {
myUri = new URI(FsConstants.VIEWFS_SCHEME, authority, "/", null, null);
fsState = new InodeTree<FileSystem>(conf, authority) {

@Override
protected
FileSystem getTargetFileSystem(final URI uri)
throws URISyntaxException, IOException {
FileSystem fs;
if (enableInnerCache) {
fs = innerCache.get(uri, config);
} else {
fs = FileSystem.get(uri, config);
protected Function<URI, FileSystem> initAndGetTargetFs() {
return new Function<URI, FileSystem>() {
@Override
public FileSystem apply(final URI uri) {
FileSystem fs;
try {
fs = ugi.doAs(new PrivilegedExceptionAction<FileSystem>() {
@Override
public FileSystem run() throws IOException {
if (enableInnerCache) {
synchronized (cache) {
return cache.get(uri, config);
}
} else {
return FileSystem.get(uri, config);
}
}
});
return new ChRootedFileSystem(fs, uri);
} catch (IOException | InterruptedException ex) {
LOG.error("Could not initialize the underlying FileSystem "
+ "object. Exception: " + ex.toString());
}
return null;
}
return new ChRootedFileSystem(fs, uri);
};
}

@Override
Expand All @@ -273,12 +291,6 @@ FileSystem getTargetFileSystem(URI[] mergeFsURIList)
}
};

if (enableInnerCache) {
// All fs instances are created and cached on startup. The cache is
// readonly after the initialize() so the concurrent access of the cache
// is safe.
cache = innerCache.unmodifiableCache();
}
workingDir = this.getHomeDirectory();
renameStrategy = RenameStrategy.valueOf(
conf.get(Constants.CONFIG_VIEWFS_RENAME_STRATEGY,
Expand Down Expand Up @@ -311,7 +323,7 @@ public ViewFileSystem(final Configuration conf) throws IOException {
this(FsConstants.VIEWFS_URI, conf);
}

public Path getTrashCanLocation(final Path f) throws FileNotFoundException {
public Path getTrashCanLocation(final Path f) throws IOException {
final InodeTree.ResolveResult<FileSystem> res =
fsState.resolve(getUriPath(f), true);
return res.isInternalDir() ? null : res.targetFileSystem.getHomeDirectory();
Expand Down Expand Up @@ -767,9 +779,34 @@ public void removeXAttr(Path path, String name) throws IOException {
public void setVerifyChecksum(final boolean verifyChecksum) {
List<InodeTree.MountPoint<FileSystem>> mountPoints =
fsState.getMountPoints();
Map<String, FileSystem> fsMap = initializeMountedFileSystems(mountPoints);
for (InodeTree.MountPoint<FileSystem> mount : mountPoints) {
fsMap.get(mount.src).setVerifyChecksum(verifyChecksum);
}
}

/**
* Initialize the target filesystem for all mount points.
* @param mountPoints The mount points
* @return Mapping of mount point and the initialized target filesystems
* @throws RuntimeException when the target file system cannot be initialized
*/
private Map<String, FileSystem> initializeMountedFileSystems(
List<InodeTree.MountPoint<FileSystem>> mountPoints) {
FileSystem fs = null;
Map<String, FileSystem> fsMap = new HashMap<>(mountPoints.size());
for (InodeTree.MountPoint<FileSystem> mount : mountPoints) {
mount.target.targetFileSystem.setVerifyChecksum(verifyChecksum);
try {
fs = mount.target.getTargetFileSystem();
fsMap.put(mount.src, fs);
} catch (IOException ex) {
String errMsg = "Not able to initialize FileSystem for mount path " +
mount.src + " with exception " + ex;
LOG.error(errMsg);
throw new RuntimeException(errMsg, ex);
}
}
return fsMap;
}

@Override
Expand All @@ -795,6 +832,9 @@ public long getDefaultBlockSize(Path f) {
return res.targetFileSystem.getDefaultBlockSize(res.remainingPath);
} catch (FileNotFoundException e) {
throw new NotInMountpointException(f, "getDefaultBlockSize");
} catch (IOException e) {
throw new RuntimeException("Not able to initialize fs in "
+ " getDefaultBlockSize for path " + f + " with exception", e);
}
}

Expand All @@ -806,6 +846,9 @@ public short getDefaultReplication(Path f) {
return res.targetFileSystem.getDefaultReplication(res.remainingPath);
} catch (FileNotFoundException e) {
throw new NotInMountpointException(f, "getDefaultReplication");
} catch (IOException e) {
throw new RuntimeException("Not able to initialize fs in "
+ " getDefaultReplication for path " + f + " with exception", e);
}
}

Expand Down Expand Up @@ -834,18 +877,20 @@ public QuotaUsage getQuotaUsage(Path f) throws IOException {
public void setWriteChecksum(final boolean writeChecksum) {
List<InodeTree.MountPoint<FileSystem>> mountPoints =
fsState.getMountPoints();
Map<String, FileSystem> fsMap = initializeMountedFileSystems(mountPoints);
for (InodeTree.MountPoint<FileSystem> mount : mountPoints) {
mount.target.targetFileSystem.setWriteChecksum(writeChecksum);
fsMap.get(mount.src).setWriteChecksum(writeChecksum);
}
}

@Override
public FileSystem[] getChildFileSystems() {
List<InodeTree.MountPoint<FileSystem>> mountPoints =
fsState.getMountPoints();
Map<String, FileSystem> fsMap = initializeMountedFileSystems(mountPoints);
Set<FileSystem> children = new HashSet<FileSystem>();
for (InodeTree.MountPoint<FileSystem> mountPoint : mountPoints) {
FileSystem targetFs = mountPoint.target.targetFileSystem;
FileSystem targetFs = fsMap.get(mountPoint.src);
children.addAll(Arrays.asList(targetFs.getChildFileSystems()));
}
return children.toArray(new FileSystem[]{});
Expand Down
42 changes: 32 additions & 10 deletions hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,12 @@

import static org.apache.hadoop.fs.viewfs.Constants.PERMISSION_555;

import com.google.common.base.Function;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
Expand Down Expand Up @@ -65,6 +67,8 @@
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
Expand Down Expand Up @@ -152,6 +156,7 @@
@InterfaceAudience.Public
@InterfaceStability.Evolving /*Evolving for a release,to be changed to Stable */
public class ViewFs extends AbstractFileSystem {
static final Logger LOG = LoggerFactory.getLogger(ViewFs.class);
final long creationTime; // of the the mount table
final UserGroupInformation ugi; // the user/group of user who created mtable
final Configuration config;
Expand Down Expand Up @@ -212,16 +217,32 @@ public ViewFs(final Configuration conf) throws IOException,
fsState = new InodeTree<AbstractFileSystem>(conf, authority) {

@Override
protected
AbstractFileSystem getTargetFileSystem(final URI uri)
throws URISyntaxException, UnsupportedFileSystemException {
String pathString = uri.getPath();
if (pathString.isEmpty()) {
pathString = "/";
protected Function<URI, AbstractFileSystem> initAndGetTargetFs() {
return new Function<URI, AbstractFileSystem>() {
@Override
public AbstractFileSystem apply(final URI uri) {
AbstractFileSystem fs;
try {
fs = ugi.doAs(
new PrivilegedExceptionAction<AbstractFileSystem>() {
@Override
public AbstractFileSystem run() throws IOException {
return AbstractFileSystem.createFileSystem(uri, config);
}
});
String pathString = uri.getPath();
if (pathString.isEmpty()) {
pathString = "/";
}
return new ChRootedFs(fs, new Path(pathString));
} catch (IOException | URISyntaxException |
InterruptedException ex) {
LOG.error("Could not initialize underlying FileSystem object"
+" for uri " + uri + "with exception: " + ex.toString());
}
return null;
}
return new ChRootedFs(
AbstractFileSystem.createFileSystem(uri, config),
new Path(pathString));
};
}

@Override
Expand Down Expand Up @@ -624,7 +645,8 @@ public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
List<Token<?>> result = new ArrayList<Token<?>>(initialListSize);
for ( int i = 0; i < mountPoints.size(); ++i ) {
List<Token<?>> tokens =
mountPoints.get(i).target.targetFileSystem.getDelegationTokens(renewer);
mountPoints.get(i).target.getTargetFileSystem()
.getDelegationTokens(renewer);
if (tokens != null) {
result.addAll(tokens);
}
Expand Down
7 changes: 3 additions & 4 deletions ...mon-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFsConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
*/
package org.apache.hadoop.fs.viewfs;

import com.google.common.base.Function;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
Expand Down Expand Up @@ -47,10 +48,8 @@ class Foo { };
new InodeTree<Foo>(conf, null) {

@Override
protected
Foo getTargetFileSystem(final URI uri)
throws URISyntaxException, UnsupportedFileSystemException {
return null;
protected Function<URI, Foo> initAndGetTargetFs() {
return null;
}

@Override
Expand Down
Loading