HDFS-6061. Allow dfs.datanode.shared.file.descriptor.path to contain multiple entries and fall back when needed (cmccabe)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1574798 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Colin McCabe 2014-03-06 08:26:26 +00:00
parent 39b70d4e1f
commit a9f2c50805
7 changed files with 123 additions and 49 deletions

View File

@ -22,11 +22,11 @@ import java.io.IOException;
import java.io.FileDescriptor; import java.io.FileDescriptor;
import org.apache.commons.lang.SystemUtils; import org.apache.commons.lang.SystemUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability; import org.apache.hadoop.classification.InterfaceStability;
import com.google.common.base.Preconditions;
/** /**
* A factory for creating shared file descriptors inside a given directory. * A factory for creating shared file descriptors inside a given directory.
* Typically, the directory will be /dev/shm or /tmp. * Typically, the directory will be /dev/shm or /tmp.
@ -45,6 +45,7 @@ import com.google.common.base.Preconditions;
@InterfaceAudience.Private @InterfaceAudience.Private
@InterfaceStability.Unstable @InterfaceStability.Unstable
public class SharedFileDescriptorFactory { public class SharedFileDescriptorFactory {
public static final Log LOG = LogFactory.getLog(SharedFileDescriptorFactory.class);
private final String prefix; private final String prefix;
private final String path; private final String path;
@ -58,18 +59,58 @@ public class SharedFileDescriptorFactory {
return null; return null;
} }
/**
* Create a new SharedFileDescriptorFactory.
*
* @param prefix The prefix to prepend to all the file names created
* by this factory.
* @param paths An array of paths to use. We will try each path in
* succession, and return a factory using the first
* usable path.
* @return The factory.
* @throws IOException If a factory could not be created for any reason.
*/
public static SharedFileDescriptorFactory create(String prefix,
String paths[]) throws IOException {
String loadingFailureReason = getLoadingFailureReason();
if (loadingFailureReason != null) {
throw new IOException(loadingFailureReason);
}
if (paths.length == 0) {
throw new IOException("no SharedFileDescriptorFactory paths were " +
"configured.");
}
StringBuilder errors = new StringBuilder();
String strPrefix = "";
for (String path : paths) {
try {
FileInputStream fis =
new FileInputStream(createDescriptor0(prefix + "test", path, 1));
fis.close();
deleteStaleTemporaryFiles0(prefix, path);
return new SharedFileDescriptorFactory(prefix, path);
} catch (IOException e) {
errors.append(strPrefix).append("Error creating file descriptor in ").
append(path).append(": ").append(e.getMessage());
strPrefix = ", ";
}
}
throw new IOException(errors.toString());
}
/** /**
* Create a SharedFileDescriptorFactory. * Create a SharedFileDescriptorFactory.
* *
* @param prefix Prefix to add to all file names we use. * @param prefix Prefix to add to all file names we use.
* @param path Path to use. * @param path Path to use.
*/ */
public SharedFileDescriptorFactory(String prefix, String path) private SharedFileDescriptorFactory(String prefix, String path) {
throws IOException {
Preconditions.checkState(getLoadingFailureReason() == null);
this.prefix = prefix; this.prefix = prefix;
this.path = path; this.path = path;
deleteStaleTemporaryFiles0(prefix, path); }
public String getPath() {
return path;
} }
/** /**

View File

@ -20,9 +20,11 @@ package org.apache.hadoop.io.nativeio;
import java.io.File; import java.io.File;
import java.io.FileInputStream; import java.io.FileInputStream;
import java.io.FileOutputStream; import java.io.FileOutputStream;
import java.io.IOException;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Assume; import org.junit.Assume;
import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.apache.commons.lang.SystemUtils; import org.apache.commons.lang.SystemUtils;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
@ -36,14 +38,19 @@ public class TestSharedFileDescriptorFactory {
private static final File TEST_BASE = private static final File TEST_BASE =
new File(System.getProperty("test.build.data", "/tmp")); new File(System.getProperty("test.build.data", "/tmp"));
@Before
public void setup() throws Exception {
Assume.assumeTrue(null ==
SharedFileDescriptorFactory.getLoadingFailureReason());
}
@Test(timeout=10000) @Test(timeout=10000)
public void testReadAndWrite() throws Exception { public void testReadAndWrite() throws Exception {
Assume.assumeTrue(NativeIO.isAvailable());
Assume.assumeTrue(SystemUtils.IS_OS_UNIX);
File path = new File(TEST_BASE, "testReadAndWrite"); File path = new File(TEST_BASE, "testReadAndWrite");
path.mkdirs(); path.mkdirs();
SharedFileDescriptorFactory factory = SharedFileDescriptorFactory factory =
new SharedFileDescriptorFactory("woot_", path.getAbsolutePath()); SharedFileDescriptorFactory.create("woot_",
new String[] { path.getAbsolutePath() });
FileInputStream inStream = FileInputStream inStream =
factory.createDescriptor("testReadAndWrite", 4096); factory.createDescriptor("testReadAndWrite", 4096);
FileOutputStream outStream = new FileOutputStream(inStream.getFD()); FileOutputStream outStream = new FileOutputStream(inStream.getFD());
@ -73,11 +80,34 @@ public class TestSharedFileDescriptorFactory {
Path.SEPARATOR + "woot2_remainder2"; Path.SEPARATOR + "woot2_remainder2";
createTempFile(remainder1); createTempFile(remainder1);
createTempFile(remainder2); createTempFile(remainder2);
new SharedFileDescriptorFactory("woot2_", path.getAbsolutePath()); SharedFileDescriptorFactory.create("woot2_",
new String[] { path.getAbsolutePath() });
// creating the SharedFileDescriptorFactory should have removed // creating the SharedFileDescriptorFactory should have removed
// the remainders // the remainders
Assert.assertFalse(new File(remainder1).exists()); Assert.assertFalse(new File(remainder1).exists());
Assert.assertFalse(new File(remainder2).exists()); Assert.assertFalse(new File(remainder2).exists());
FileUtil.fullyDelete(path); FileUtil.fullyDelete(path);
} }
@Test(timeout=60000)
public void testDirectoryFallbacks() throws Exception {
File nonExistentPath = new File(TEST_BASE, "nonexistent");
File permissionDeniedPath = new File("/");
File goodPath = new File(TEST_BASE, "testDirectoryFallbacks");
goodPath.mkdirs();
try {
SharedFileDescriptorFactory.create("shm_",
new String[] { nonExistentPath.getAbsolutePath(),
permissionDeniedPath.getAbsolutePath() });
Assert.fail();
} catch (IOException e) {
}
SharedFileDescriptorFactory factory =
SharedFileDescriptorFactory.create("shm_",
new String[] { nonExistentPath.getAbsolutePath(),
permissionDeniedPath.getAbsolutePath(),
goodPath.getAbsolutePath() } );
Assert.assertEquals(goodPath.getAbsolutePath(), factory.getPath());
FileUtil.fullyDelete(goodPath);
}
} }

View File

@ -138,6 +138,9 @@ Release 2.4.0 - UNRELEASED
HDFS-6044. Add property for setting the NFS look up time for users HDFS-6044. Add property for setting the NFS look up time for users
(brandonli) (brandonli)
HDFS-6061. Allow dfs.datanode.shared.file.descriptor.path to contain
multiple entries and fall back when needed (cmccabe)
OPTIMIZATIONS OPTIMIZATIONS
HDFS-5790. LeaseManager.findPath is very slow when many leases need recovery HDFS-5790. LeaseManager.findPath is very slow when many leases need recovery

View File

@ -471,8 +471,8 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
public static final String DFS_NAMENODE_STARTUP_KEY = "dfs.namenode.startup"; public static final String DFS_NAMENODE_STARTUP_KEY = "dfs.namenode.startup";
public static final String DFS_DATANODE_KEYTAB_FILE_KEY = "dfs.datanode.keytab.file"; public static final String DFS_DATANODE_KEYTAB_FILE_KEY = "dfs.datanode.keytab.file";
public static final String DFS_DATANODE_USER_NAME_KEY = "dfs.datanode.kerberos.principal"; public static final String DFS_DATANODE_USER_NAME_KEY = "dfs.datanode.kerberos.principal";
public static final String DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATH = "dfs.datanode.shared.file.descriptor.path"; public static final String DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATHS = "dfs.datanode.shared.file.descriptor.paths";
public static final String DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATH_DEFAULT = "/dev/shm"; public static final String DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATHS_DEFAULT = "/dev/shm,/tmp";
public static final String DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS = "dfs.short.circuit.shared.memory.watcher.interrupt.check.ms"; public static final String DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS = "dfs.short.circuit.shared.memory.watcher.interrupt.check.ms";
public static final int DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS_DEFAULT = 60000; public static final int DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS_DEFAULT = 60000;
public static final String DFS_NAMENODE_KEYTAB_FILE_KEY = "dfs.namenode.keytab.file"; public static final String DFS_NAMENODE_KEYTAB_FILE_KEY = "dfs.namenode.keytab.file";

View File

@ -17,14 +17,15 @@
*/ */
package org.apache.hadoop.hdfs.server.datanode; package org.apache.hadoop.hdfs.server.datanode;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATH; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATHS;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATH_DEFAULT; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATHS_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS_DEFAULT; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS_DEFAULT;
import java.io.Closeable; import java.io.Closeable;
import java.io.FileInputStream; import java.io.FileInputStream;
import java.io.IOException; import java.io.IOException;
import java.util.Collections;
import java.util.HashMap; import java.util.HashMap;
import java.util.Iterator; import java.util.Iterator;
import java.util.Set; import java.util.Set;
@ -45,7 +46,9 @@ import org.apache.hadoop.net.unix.DomainSocket;
import org.apache.hadoop.net.unix.DomainSocketWatcher; import org.apache.hadoop.net.unix.DomainSocketWatcher;
import com.google.common.base.Preconditions; import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.HashMultimap; import com.google.common.collect.HashMultimap;
import com.google.common.collect.Iterables;
/* /*
* Manages client short-circuit memory segments on the DataNode. * Manages client short-circuit memory segments on the DataNode.
@ -149,38 +152,35 @@ public class ShortCircuitRegistry {
SharedFileDescriptorFactory shmFactory = null; SharedFileDescriptorFactory shmFactory = null;
DomainSocketWatcher watcher = null; DomainSocketWatcher watcher = null;
try { try {
String loadingFailureReason =
SharedFileDescriptorFactory.getLoadingFailureReason();
if (loadingFailureReason != null) {
if (LOG.isDebugEnabled()) {
LOG.debug("Disabling ShortCircuitRegistry because " +
loadingFailureReason);
}
return;
}
String shmPath = conf.get(DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATH,
DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATH_DEFAULT);
if (shmPath.isEmpty()) {
LOG.debug("Disabling ShortCircuitRegistry because shmPath was not set.");
return;
}
int interruptCheck = conf.getInt( int interruptCheck = conf.getInt(
DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS, DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS,
DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS_DEFAULT); DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS_DEFAULT);
if (interruptCheck <= 0) { if (interruptCheck <= 0) {
if (LOG.isDebugEnabled()) { throw new IOException(
LOG.debug("Disabling ShortCircuitRegistry because " + DFS_SHORT_CIRCUIT_SHARED_MEMORY_WATCHER_INTERRUPT_CHECK_MS +
"interruptCheckMs was set to " + interruptCheck); " was set to " + interruptCheck);
} }
return; String shmPaths[] =
conf.getTrimmedStrings(DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATHS);
if (shmPaths.length == 0) {
shmPaths =
DFS_DATANODE_SHARED_FILE_DESCRIPTOR_PATHS_DEFAULT.split(",");
}
shmFactory = SharedFileDescriptorFactory.
create("HadoopShortCircuitShm_", shmPaths);
String dswLoadingFailure = DomainSocketWatcher.getLoadingFailureReason();
if (dswLoadingFailure != null) {
throw new IOException(dswLoadingFailure);
} }
shmFactory =
new SharedFileDescriptorFactory("HadoopShortCircuitShm_", shmPath);
watcher = new DomainSocketWatcher(interruptCheck); watcher = new DomainSocketWatcher(interruptCheck);
enabled = true; enabled = true;
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("created new ShortCircuitRegistry with interruptCheck=" + LOG.debug("created new ShortCircuitRegistry with interruptCheck=" +
interruptCheck + ", shmPath=" + shmPath); interruptCheck + ", shmPath=" + shmFactory.getPath());
}
} catch (IOException e) {
if (LOG.isDebugEnabled()) {
LOG.debug("Disabling ShortCircuitRegistry", e);
} }
} finally { } finally {
this.enabled = enabled; this.enabled = enabled;

View File

@ -1138,13 +1138,13 @@
</property> </property>
<property> <property>
<name>dfs.datanode.shared.file.descriptor.path</name> <name>dfs.datanode.shared.file.descriptor.paths</name>
<value>/dev/shm</value> <value>/dev/shm,/tmp</value>
<description> <description>
The path to use when creating file descriptors that will be shared A comma-separated list of paths to use when creating file descriptors that
between the DataNode and the DFSClient. Typically we use /dev/shm, so will be shared between the DataNode and the DFSClient. Typically we use
that the file descriptors will not be written to disk. Systems that /dev/shm, so that the file descriptors will not be written to disk.
don't have /dev/shm should use /tmp. Systems that don't have /dev/shm will fall back to /tmp by default.
</description> </description>
</property> </property>

View File

@ -22,11 +22,9 @@ import java.io.FileInputStream;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Iterator; import java.util.Iterator;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileUtil; import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.io.nativeio.NativeIO;
import org.apache.hadoop.io.nativeio.SharedFileDescriptorFactory; import org.apache.hadoop.io.nativeio.SharedFileDescriptorFactory;
import org.apache.hadoop.hdfs.ExtendedBlockId; import org.apache.hadoop.hdfs.ExtendedBlockId;
import org.apache.hadoop.hdfs.ShortCircuitShm; import org.apache.hadoop.hdfs.ShortCircuitShm;
@ -45,8 +43,8 @@ public class TestShortCircuitShm {
@Before @Before
public void before() { public void before() {
Assume.assumeTrue(NativeIO.isAvailable()); Assume.assumeTrue(null ==
Assume.assumeTrue(SystemUtils.IS_OS_UNIX); SharedFileDescriptorFactory.getLoadingFailureReason());
} }
@Test(timeout=60000) @Test(timeout=60000)
@ -54,7 +52,8 @@ public class TestShortCircuitShm {
File path = new File(TEST_BASE, "testStartupShutdown"); File path = new File(TEST_BASE, "testStartupShutdown");
path.mkdirs(); path.mkdirs();
SharedFileDescriptorFactory factory = SharedFileDescriptorFactory factory =
new SharedFileDescriptorFactory("shm_", path.getAbsolutePath()); SharedFileDescriptorFactory.create("shm_",
new String[] { path.getAbsolutePath() } );
FileInputStream stream = FileInputStream stream =
factory.createDescriptor("testStartupShutdown", 4096); factory.createDescriptor("testStartupShutdown", 4096);
ShortCircuitShm shm = new ShortCircuitShm(ShmId.createRandom(), stream); ShortCircuitShm shm = new ShortCircuitShm(ShmId.createRandom(), stream);
@ -68,7 +67,8 @@ public class TestShortCircuitShm {
File path = new File(TEST_BASE, "testAllocateSlots"); File path = new File(TEST_BASE, "testAllocateSlots");
path.mkdirs(); path.mkdirs();
SharedFileDescriptorFactory factory = SharedFileDescriptorFactory factory =
new SharedFileDescriptorFactory("shm_", path.getAbsolutePath()); SharedFileDescriptorFactory.create("shm_",
new String[] { path.getAbsolutePath() });
FileInputStream stream = FileInputStream stream =
factory.createDescriptor("testAllocateSlots", 4096); factory.createDescriptor("testAllocateSlots", 4096);
ShortCircuitShm shm = new ShortCircuitShm(ShmId.createRandom(), stream); ShortCircuitShm shm = new ShortCircuitShm(ShmId.createRandom(), stream);