Skip to content

Commit

Permalink
HDFS-8152. Refactoring of lazy persist storage cases. (Arpit Agarwal)
Browse files Browse the repository at this point in the history
  • Loading branch information
@arp7
arp7 committed Apr 19, 2015
1 parent 5459b24 commit 8511d80
Show file tree
Hide file tree
Showing 8 changed files with 152 additions and 108 deletions.
2 changes: 2 additions & 0 deletions hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -448,6 +448,8 @@ Release 2.8.0 - UNRELEASED
HDFS-8165. Move GRANDFATHER_GENERATION_STAMP and GRANDFATER_INODE_ID to
hdfs-client. (wheat9)

HDFS-8152. Refactoring of lazy persist storage cases. (Arpit Agarwal)

OPTIMIZATIONS

HDFS-8026. Trace FSOutputSummer#writeChecksumChunks rather than
Expand Down
152 changes: 89 additions & 63 deletions .../test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/LazyPersistTestCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
*/
package org.apache.hadoop.hdfs.server.datanode.fsdataset.impl;

import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;

import static org.apache.hadoop.fs.CreateFlag.CREATE;
import static org.apache.hadoop.fs.CreateFlag.LAZY_PERSIST;
import static org.apache.hadoop.fs.StorageType.DEFAULT;
Expand Down Expand Up @@ -45,6 +47,7 @@
import java.util.Set;
import java.util.UUID;

import com.google.common.base.Preconditions;
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
Expand Down Expand Up @@ -228,11 +231,15 @@ protected final void makeTestFile(Path path, long length,
* If ramDiskStorageLimit is >=0, then RAM_DISK capacity is artificially
* capped. If ramDiskStorageLimit < 0 then it is ignored.
*/
protected final void startUpCluster(boolean hasTransientStorage,
final int ramDiskReplicaCapacity,
final boolean useSCR,
final boolean useLegacyBlockReaderLocal)
throws IOException {
protected final void startUpCluster(
int numDatanodes,
boolean hasTransientStorage,
StorageType[] storageTypes,
int ramDiskReplicaCapacity,
long ramDiskStorageLimit,
long evictionLowWatermarkReplicas,
boolean useSCR,
boolean useLegacyBlockReaderLocal) throws IOException {

Configuration conf = new Configuration();
conf.setLong(DFS_BLOCK_SIZE_KEY, BLOCK_SIZE);
Expand All @@ -243,40 +250,47 @@ protected final void startUpCluster(boolean hasTransientStorage,
HEARTBEAT_RECHECK_INTERVAL_MSEC);
conf.setInt(DFS_DATANODE_LAZY_WRITER_INTERVAL_SEC,
LAZY_WRITER_INTERVAL_SEC);
conf.setInt(DFS_DATANODE_RAM_DISK_LOW_WATERMARK_BYTES,
EVICTION_LOW_WATERMARK * BLOCK_SIZE);
conf.setLong(DFS_DATANODE_RAM_DISK_LOW_WATERMARK_BYTES,
evictionLowWatermarkReplicas * BLOCK_SIZE);

if (useSCR) {
conf.setBoolean(HdfsClientConfigKeys.Read.ShortCircuit.KEY, true);
// Do not share a client context across tests.
conf.set(DFS_CLIENT_CONTEXT, UUID.randomUUID().toString());
conf.set(DFS_BLOCK_LOCAL_PATH_ACCESS_USER_KEY,
UserGroupInformation.getCurrentUser().getShortUserName());
if (useLegacyBlockReaderLocal) {
conf.setBoolean(DFS_CLIENT_USE_LEGACY_BLOCKREADERLOCAL, true);
conf.set(DFS_BLOCK_LOCAL_PATH_ACCESS_USER_KEY,
UserGroupInformation.getCurrentUser().getShortUserName());
} else {
sockDir = new TemporarySocketDirectory();
conf.set(DFS_DOMAIN_SOCKET_PATH_KEY, new File(sockDir.getDir(),
this.getClass().getSimpleName() + "._PORT.sock").getAbsolutePath());
}
}

long[] capacities = null;
Preconditions.checkState(
ramDiskReplicaCapacity < 0 || ramDiskStorageLimit < 0,
"Cannot specify non-default values for both ramDiskReplicaCapacity "
+ "and ramDiskStorageLimit");

long[] capacities;
if (hasTransientStorage && ramDiskReplicaCapacity >= 0) {
// Convert replica count to byte count, add some delta for .meta and
// VERSION files.
long ramDiskStorageLimit = ((long) ramDiskReplicaCapacity * BLOCK_SIZE) +
ramDiskStorageLimit = ((long) ramDiskReplicaCapacity * BLOCK_SIZE) +
(BLOCK_SIZE - 1);
capacities = new long[] { ramDiskStorageLimit, -1 };
}
capacities = new long[] { ramDiskStorageLimit, -1 };

cluster = new MiniDFSCluster
.Builder(conf)
.numDataNodes(REPL_FACTOR)
.numDataNodes(numDatanodes)
.storageCapacities(capacities)
.storageTypes(hasTransientStorage ?
new StorageType[]{ RAM_DISK, DEFAULT } : null)
.storageTypes(storageTypes != null ? storageTypes :
(hasTransientStorage ? new StorageType[]{RAM_DISK, DEFAULT} : null))
.build();
cluster.waitActive();

fs = cluster.getFileSystem();
client = fs.getClient();
try {
Expand All @@ -287,65 +301,77 @@ protected final void startUpCluster(boolean hasTransientStorage,
LOG.info("Cluster startup complete");
}

ClusterWithRamDiskBuilder getClusterBuilder() {
return new ClusterWithRamDiskBuilder();
}

/**
* If ramDiskStorageLimit is >=0, then RAM_DISK capacity is artificially
* capped. If ramDiskStorageLimit < 0 then it is ignored.
* Builder class that allows controlling RAM disk-specific properties for a
* MiniDFSCluster.
*/
protected final void startUpCluster(final int numDataNodes,
final StorageType[] storageTypes,
final long ramDiskStorageLimit,
final boolean useSCR)
throws IOException {
class ClusterWithRamDiskBuilder {
public ClusterWithRamDiskBuilder setNumDatanodes(
int numDatanodes) {
this.numDatanodes = numDatanodes;
return this;
}

Configuration conf = new Configuration();
conf.setLong(DFS_BLOCK_SIZE_KEY, BLOCK_SIZE);
conf.setInt(DFS_NAMENODE_LAZY_PERSIST_FILE_SCRUB_INTERVAL_SEC,
LAZY_WRITE_FILE_SCRUBBER_INTERVAL_SEC);
conf.setLong(DFS_HEARTBEAT_INTERVAL_KEY, HEARTBEAT_INTERVAL_SEC);
conf.setInt(DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
HEARTBEAT_RECHECK_INTERVAL_MSEC);
conf.setInt(DFS_DATANODE_LAZY_WRITER_INTERVAL_SEC,
LAZY_WRITER_INTERVAL_SEC);
public ClusterWithRamDiskBuilder setStorageTypes(
StorageType[] storageTypes) {
this.storageTypes = storageTypes;
return this;
}

if (useSCR)
{
conf.setBoolean(HdfsClientConfigKeys.Read.ShortCircuit.KEY,useSCR);
conf.set(DFS_CLIENT_CONTEXT, UUID.randomUUID().toString());
sockDir = new TemporarySocketDirectory();
conf.set(DFS_DOMAIN_SOCKET_PATH_KEY, new File(sockDir.getDir(),
this.getClass().getSimpleName() + "._PORT.sock").getAbsolutePath());
conf.set(DFS_BLOCK_LOCAL_PATH_ACCESS_USER_KEY,
UserGroupInformation.getCurrentUser().getShortUserName());
public ClusterWithRamDiskBuilder setRamDiskReplicaCapacity(
int ramDiskReplicaCapacity) {
this.ramDiskReplicaCapacity = ramDiskReplicaCapacity;
return this;
}

cluster = new MiniDFSCluster
.Builder(conf)
.numDataNodes(numDataNodes)
.storageTypes(storageTypes != null ?
storageTypes : new StorageType[] { DEFAULT, DEFAULT })
.build();
fs = cluster.getFileSystem();
client = fs.getClient();
public ClusterWithRamDiskBuilder setRamDiskStorageLimit(
long ramDiskStorageLimit) {
this.ramDiskStorageLimit = ramDiskStorageLimit;
return this;
}

// Artificially cap the storage capacity of the RAM_DISK volume.
if (ramDiskStorageLimit >= 0) {
List<? extends FsVolumeSpi> volumes =
cluster.getDataNodes().get(0).getFSDataset().getVolumes();
public ClusterWithRamDiskBuilder setUseScr(boolean useScr) {
this.useScr = useScr;
return this;
}

for (FsVolumeSpi volume : volumes) {
if (volume.getStorageType() == RAM_DISK) {
((FsVolumeImpl) volume).setCapacityForTesting(ramDiskStorageLimit);
}
}
public ClusterWithRamDiskBuilder setHasTransientStorage(
boolean hasTransientStorage) {
this.hasTransientStorage = hasTransientStorage;
return this;
}

LOG.info("Cluster startup complete");
}
public ClusterWithRamDiskBuilder setUseLegacyBlockReaderLocal(
boolean useLegacyBlockReaderLocal) {
this.useLegacyBlockReaderLocal = useLegacyBlockReaderLocal;
return this;
}

public ClusterWithRamDiskBuilder setEvictionLowWatermarkReplicas(
long evictionLowWatermarkReplicas) {
this.evictionLowWatermarkReplicas = evictionLowWatermarkReplicas;
return this;
}

public void build() throws IOException {
LazyPersistTestCase.this.startUpCluster(
numDatanodes, hasTransientStorage, storageTypes, ramDiskReplicaCapacity,
ramDiskStorageLimit, evictionLowWatermarkReplicas,
useScr, useLegacyBlockReaderLocal);
}

protected final void startUpCluster(boolean hasTransientStorage,
final int ramDiskReplicaCapacity)
throws IOException {
startUpCluster(hasTransientStorage, ramDiskReplicaCapacity, false, false);
private int numDatanodes = REPL_FACTOR;
private StorageType[] storageTypes = null;
private int ramDiskReplicaCapacity = -1;
private long ramDiskStorageLimit = -1;
private boolean hasTransientStorage = true;
private boolean useScr = false;
private boolean useLegacyBlockReaderLocal = false;
private long evictionLowWatermarkReplicas = EVICTION_LOW_WATERMARK;
}

protected final void triggerBlockReport()
Expand Down
20 changes: 5 additions & 15 deletions ...test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestLazyPersistFiles.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,6 @@
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DFSTestUtil;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
import org.apache.hadoop.test.GenericTestUtils;
import org.junit.Assert;
import org.junit.Test;
Expand All @@ -34,17 +29,12 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.ArrayList;
import java.util.Collections;

import static org.apache.hadoop.fs.StorageType.DEFAULT;
import static org.apache.hadoop.fs.StorageType.RAM_DISK;
import static org.apache.hadoop.hdfs.DFSConfigKeys.*;

import static org.hamcrest.core.Is.is;
import static org.hamcrest.core.IsNot.not;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

public class TestLazyPersistFiles extends LazyPersistTestCase {
Expand All @@ -56,7 +46,7 @@ public class TestLazyPersistFiles extends LazyPersistTestCase {
*/
@Test
public void testAppendIsDenied() throws IOException {
startUpCluster(true, -1);
getClusterBuilder().build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand All @@ -77,7 +67,7 @@ public void testAppendIsDenied() throws IOException {
*/
@Test
public void testTruncateIsDenied() throws IOException {
startUpCluster(true, -1);
getClusterBuilder().build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand All @@ -99,7 +89,7 @@ public void testTruncateIsDenied() throws IOException {
@Test
public void testCorruptFilesAreDiscarded()
throws IOException, InterruptedException {
startUpCluster(true, 2);
getClusterBuilder().setRamDiskReplicaCapacity(2).build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path1 = new Path("/" + METHOD_NAME + ".01.dat");

Expand Down Expand Up @@ -136,7 +126,7 @@ public void testCorruptFilesAreDiscarded()
@Test
public void testConcurrentRead()
throws Exception {
startUpCluster(true, 2);
getClusterBuilder().setRamDiskReplicaCapacity(2).build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
final Path path1 = new Path("/" + METHOD_NAME + ".dat");

Expand Down Expand Up @@ -187,7 +177,7 @@ public void run() {
@Test
public void testConcurrentWrites()
throws IOException, InterruptedException {
startUpCluster(true, 9);
getClusterBuilder().setRamDiskReplicaCapacity(9).build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
final int SEED = 0xFADED;
final int NUM_WRITERS = 4;
Expand Down
8 changes: 4 additions & 4 deletions ...est/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestLazyPersistPolicy.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
public class TestLazyPersistPolicy extends LazyPersistTestCase {
@Test
public void testPolicyNotSetByDefault() throws IOException {
startUpCluster(false, -1);
getClusterBuilder().build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand All @@ -47,7 +47,7 @@ public void testPolicyNotSetByDefault() throws IOException {

@Test
public void testPolicyPropagation() throws IOException {
startUpCluster(false, -1);
getClusterBuilder().build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand All @@ -59,7 +59,7 @@ public void testPolicyPropagation() throws IOException {

@Test
public void testPolicyPersistenceInEditLog() throws IOException {
startUpCluster(false, -1);
getClusterBuilder().build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand All @@ -73,7 +73,7 @@ public void testPolicyPersistenceInEditLog() throws IOException {

@Test
public void testPolicyPersistenceInFsImage() throws IOException {
startUpCluster(false, -1);
getClusterBuilder().build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand Down
12 changes: 6 additions & 6 deletions ...rg/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestLazyPersistReplicaPlacement.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
public class TestLazyPersistReplicaPlacement extends LazyPersistTestCase {
@Test
public void testPlacementOnRamDisk() throws IOException {
startUpCluster(true, -1);
getClusterBuilder().build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand All @@ -43,7 +43,7 @@ public void testPlacementOnRamDisk() throws IOException {

@Test
public void testPlacementOnSizeLimitedRamDisk() throws IOException {
startUpCluster(true, 3);
getClusterBuilder().setRamDiskReplicaCapacity(3).build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path1 = new Path("/" + METHOD_NAME + ".01.dat");
Path path2 = new Path("/" + METHOD_NAME + ".02.dat");
Expand All @@ -62,7 +62,7 @@ public void testPlacementOnSizeLimitedRamDisk() throws IOException {
*/
@Test
public void testFallbackToDisk() throws IOException {
startUpCluster(false, -1);
getClusterBuilder().setHasTransientStorage(false).build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand All @@ -76,7 +76,7 @@ public void testFallbackToDisk() throws IOException {
*/
@Test
public void testFallbackToDiskFull() throws Exception {
startUpCluster(false, 0);
getClusterBuilder().setRamDiskReplicaCapacity(0).build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand All @@ -95,7 +95,7 @@ public void testFallbackToDiskFull() throws Exception {
@Test
public void testFallbackToDiskPartial()
throws IOException, InterruptedException {
startUpCluster(true, 2);
getClusterBuilder().setRamDiskReplicaCapacity(2).build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand Down Expand Up @@ -134,7 +134,7 @@ public void testFallbackToDiskPartial()
*/
@Test
public void testRamDiskNotChosenByDefault() throws IOException {
startUpCluster(true, -1);
getClusterBuilder().build();
final String METHOD_NAME = GenericTestUtils.getMethodName();
Path path = new Path("/" + METHOD_NAME + ".dat");

Expand Down
Loading

0 comments on commit 8511d80

Please sign in to comment.