Skip to content

HDFS-15963. Unreleased volume references cause an infinite loop. #2889

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

Merged
Hexiaoqiao merged 5 commits into from
Apr 16, 2021
Merged

HDFS-15963. Unreleased volume references cause an infinite loop. #2889

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fa37c6a
HDFS-15963. Unreleased volume references cause an infinite loop.
Apr 10, 2021
f5215c8
revise based on review
Apr 12, 2021
08b8eb0
small changes
Apr 13, 2021
68dd50e
bug fixed
Apr 15, 2021
28864d4
add timeout
Apr 15, 2021
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
1 change: 1 addition & 0 deletions ...project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockSender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -432,6 +432,7 @@ class BlockSender implements java.io.Closeable {
ris = new ReplicaInputStreams(
blockIn, checksumIn, volumeRef, fileIoProvider);
} catch (IOException ioe) {
IOUtils.cleanupWithLogger(null, volumeRef);
IOUtils.closeStream(this);
org.apache.commons.io.IOUtils.closeQuietly(blockIn);
org.apache.commons.io.IOUtils.closeQuietly(checksumIn);
Expand Down
71 changes: 41 additions & 30 deletions ...java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetAsyncDiskService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,18 +167,26 @@ synchronized long countPendingDeletions() {
* Execute the task sometime in the future, using ThreadPools.
*/
synchronized void execute(FsVolumeImpl volume, Runnable task) {
if (executors == null) {
throw new RuntimeException("AsyncDiskService is already shutdown");
}
if (volume == null) {
throw new RuntimeException("A null volume does not have a executor");
}
ThreadPoolExecutor executor = executors.get(volume.getStorageID());
if (executor == null) {
throw new RuntimeException("Cannot find volume " + volume
+ " for execution of task " + task);
} else {
executor.execute(task);
try {
if (executors == null) {
throw new RuntimeException("AsyncDiskService is already shutdown");
}
if (volume == null) {
throw new RuntimeException("A null volume does not have a executor");
}
ThreadPoolExecutor executor = executors.get(volume.getStorageID());
if (executor == null) {
throw new RuntimeException("Cannot find volume " + volume
+ " for execution of task " + task);
} else {
executor.execute(task);
}
} catch (RuntimeException re) {
if (task instanceof ReplicaFileDeleteTask) {
IOUtils.cleanupWithLogger(null,
((ReplicaFileDeleteTask) task).volumeRef);
}
throw re;
}
}

Expand Down Expand Up @@ -314,28 +322,31 @@ private boolean moveFiles() {

@Override
public void run() {
final long blockLength = replicaToDelete.getBlockDataLength();
final long metaLength = replicaToDelete.getMetadataLength();
boolean result;
try {
final long blockLength = replicaToDelete.getBlockDataLength();
final long metaLength = replicaToDelete.getMetadataLength();
boolean result;

result = (trashDirectory == null) ? deleteFiles() : moveFiles();
result = (trashDirectory == null) ? deleteFiles() : moveFiles();

if (!result) {
LOG.warn("Unexpected error trying to "
+ (trashDirectory == null ? "delete" : "move")
+ " block " + block.getBlockPoolId() + " " + block.getLocalBlock()
+ " at file " + replicaToDelete.getBlockURI() + ". Ignored.");
} else {
if(block.getLocalBlock().getNumBytes() != BlockCommand.NO_ACK){
datanode.notifyNamenodeDeletedBlock(block, volume.getStorageID());
if (!result) {
LOG.warn("Unexpected error trying to "
+ (trashDirectory == null ? "delete" : "move")
+ " block " + block.getBlockPoolId() + " " + block.getLocalBlock()
+ " at file " + replicaToDelete.getBlockURI() + ". Ignored.");
} else {
if (block.getLocalBlock().getNumBytes() != BlockCommand.NO_ACK) {
datanode.notifyNamenodeDeletedBlock(block, volume.getStorageID());
}
volume.onBlockFileDeletion(block.getBlockPoolId(), blockLength);
volume.onMetaFileDeletion(block.getBlockPoolId(), metaLength);
LOG.info("Deleted " + block.getBlockPoolId() + " " +
block.getLocalBlock() + " URI " + replicaToDelete.getBlockURI());
}
volume.onBlockFileDeletion(block.getBlockPoolId(), blockLength);
volume.onMetaFileDeletion(block.getBlockPoolId(), metaLength);
LOG.info("Deleted " + block.getBlockPoolId() + " "
+ block.getLocalBlock() + " URI " + replicaToDelete.getBlockURI());
updateDeletedBlockId(block);
} finally {
IOUtils.cleanupWithLogger(null, this.volumeRef);
}
updateDeletedBlockId(block);
IOUtils.cleanupWithLogger(null, volumeRef);
}
}

Expand Down
2 changes: 1 addition & 1 deletion ...dfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,7 @@ private void checkReference() {
}

@VisibleForTesting
int getReferenceCount() {
public int getReferenceCount() {
return this.reference.getReferenceCount();
}

Expand Down
29 changes: 19 additions & 10 deletions ...org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/RamDiskAsyncLazyPersistService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package org.apache.hadoop.hdfs.server.datanode.fsdataset.impl;

import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
Expand Down Expand Up @@ -153,16 +154,24 @@ synchronized boolean queryVolume(FsVolumeImpl volume) {
* Execute the task sometime in the future, using ThreadPools.
*/
synchronized void execute(String storageId, Runnable task) {
if (executors == null) {
throw new RuntimeException(
"AsyncLazyPersistService is already shutdown");
}
ThreadPoolExecutor executor = executors.get(storageId);
if (executor == null) {
throw new RuntimeException("Cannot find root storage volume with id " +
storageId + " for execution of task " + task);
} else {
executor.execute(task);
try {
if (executors == null) {
throw new RuntimeException(
"AsyncLazyPersistService is already shutdown");
}
ThreadPoolExecutor executor = executors.get(storageId);
if (executor == null) {
throw new RuntimeException("Cannot find root storage volume with id " +
storageId + " for execution of task " + task);
} else {
executor.execute(task);
}
} catch (RuntimeException re) {
if (task instanceof ReplicaLazyPersistTask) {
IOUtils.cleanupWithLogger(null,
((ReplicaLazyPersistTask) task).targetVolume);
}
throw re;
}
}

Expand Down
55 changes: 55 additions & 0 deletions ...fs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDataTransferProtocol.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,9 @@
import java.nio.ByteBuffer;
import java.util.Random;

import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsVolumeImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
Expand Down Expand Up @@ -562,4 +565,56 @@ void writeBlock(ExtendedBlock block, BlockConstructionStage stage,
checksum, CachingStrategy.newDefaultStrategy(), false, false,
null, null, new String[0]);
}

@Test(timeout = 30000)
public void testReleaseVolumeRefIfExceptionThrown()
throws IOException, InterruptedException {
Path file = new Path("dataprotocol.dat");
int numDataNodes = 1;

Configuration conf = new HdfsConfiguration();
conf.setInt(DFSConfigKeys.DFS_REPLICATION_KEY, numDataNodes);
MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(
numDataNodes).build();
try {
cluster.waitActive();
datanode = cluster.getFileSystem().getDataNodeStats(
DatanodeReportType.LIVE)[0];
dnAddr = NetUtils.createSocketAddr(datanode.getXferAddr());
FileSystem fileSys = cluster.getFileSystem();

int fileLen = Math.min(
conf.getInt(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 4096), 4096);

DFSTestUtil.createFile(fileSys, file, fileLen, fileLen,
fileSys.getDefaultBlockSize(file),
fileSys.getDefaultReplication(file), 0L);

// Get the first blockid for the file.
final ExtendedBlock firstBlock = DFSTestUtil.getFirstBlock(fileSys, file);

String bpid = cluster.getNamesystem().getBlockPoolId();
ExtendedBlock blk = new ExtendedBlock(bpid, firstBlock.getLocalBlock());
sendBuf.reset();
recvBuf.reset();

// Delete the meta file to create a exception in BlockSender constructor.
DataNode dn = cluster.getDataNodes().get(0);
cluster.getMaterializedReplica(0, blk).deleteMeta();

FsVolumeImpl volume = (FsVolumeImpl) DataNodeTestUtils.getFSDataset(
dn).getVolume(blk);
int beforeCnt = volume.getReferenceCount();

sender.copyBlock(blk, BlockTokenSecretManager.DUMMY_TOKEN);
sendRecvData("Copy a block.", false);
Thread.sleep(3000);

int afterCnt = volume.getReferenceCount();
assertEquals(beforeCnt, afterCnt);

} finally {
cluster.shutdown();
}
}
}
34 changes: 34 additions & 0 deletions ...rc/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsDatasetImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.hadoop.fs.DF;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
import org.apache.hadoop.hdfs.server.datanode.DirectoryScanner;
import org.apache.hadoop.hdfs.server.datanode.LocalReplica;
import org.apache.hadoop.thirdparty.com.google.common.collect.Lists;

import java.io.OutputStream;
Expand Down Expand Up @@ -1805,4 +1806,37 @@ public void testNotifyNamenodeMissingOrNewBlock() throws Exception {
cluster.shutdown();
}
}

@Test(timeout = 20000)
public void testReleaseVolumeRefIfExceptionThrown() throws IOException {
MiniDFSCluster cluster = new MiniDFSCluster.Builder(
new HdfsConfiguration()).build();
cluster.waitActive();
FsVolumeImpl vol = (FsVolumeImpl) dataset.getFsVolumeReferences().get(0);
ExtendedBlock eb;
ReplicaInfo info;
int beforeCnt = 0;
try {
List<Block> blockList = new ArrayList<Block>();
eb = new ExtendedBlock(BLOCKPOOL, 1, 1, 1001);
info = new FinalizedReplica(
eb.getLocalBlock(), vol, vol.getCurrentDir().getParentFile());
dataset.volumeMap.add(BLOCKPOOL, info);
((LocalReplica) info).getBlockFile().createNewFile();
((LocalReplica) info).getMetaFile().createNewFile();
blockList.add(info);

// Create a runtime exception.
dataset.asyncDiskService.shutdown();

beforeCnt = vol.getReferenceCount();
dataset.invalidate(BLOCKPOOL, blockList.toArray(new Block[0]));

} catch (RuntimeException re) {
int afterCnt = vol.getReferenceCount();
assertEquals(beforeCnt, afterCnt);
} finally {
cluster.shutdown();
}
}
}
37 changes: 37 additions & 0 deletions ...test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestLazyPersistFiles.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@
package org.apache.hadoop.hdfs.server.datanode.fsdataset.impl;
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.ThreadUtil;
import org.junit.Assert;
Expand Down Expand Up @@ -280,4 +283,38 @@ public void run() {
}
}
}

@Test(timeout = 20000)
public void testReleaseVolumeRefIfExceptionThrown()
throws IOException, InterruptedException {
getClusterBuilder().setRamDiskReplicaCapacity(2).build();
final String methodName = GenericTestUtils.getMethodName();
final int seed = 0xFADED;
Path path = new Path("/" + methodName + ".Writer.File.dat");

DataNode dn = cluster.getDataNodes().get(0);
FsDatasetSpi.FsVolumeReferences volumes =
DataNodeTestUtils.getFSDataset(dn).getFsVolumeReferences();
int[] beforeCnts = new int[volumes.size()];
FsDatasetImpl ds = (FsDatasetImpl) DataNodeTestUtils.getFSDataset(dn);

// Create a runtime exception.
ds.asyncLazyPersistService.shutdown();
for (int i = 0; i < volumes.size(); ++i) {
beforeCnts[i] = ((FsVolumeImpl) volumes.get(i)).getReferenceCount();
}

makeRandomTestFile(path, BLOCK_SIZE, true, seed);
Thread.sleep(3 * LAZY_WRITER_INTERVAL_SEC * 1000);

for (int i = 0; i < volumes.size(); ++i) {
int afterCnt = ((FsVolumeImpl) volumes.get(i)).getReferenceCount();
// LazyWriter keeps trying to save copies even if
// asyncLazyPersistService is already shutdown.
// If we do not release references, the number of
// references will increase infinitely.
Assert.assertTrue(
beforeCnts[i] == afterCnt || beforeCnts[i] == (afterCnt - 1));
}
}
}