big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-12523. Thread pools in ErasureCodingWorker do not shutdown. (Huafeng Wang via Lei)

Browse Source
This commit is contained in:
Lei Xu 2017-09-21 16:10:32 -07:00
parent 9d3e4cccf9
commit 1267ff22ce
3 changed files with 17 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
View File

@ -2077,6 +2077,10 @@ public void shutdown() {
ipcServer.stop(); ipcServer.stop();
} }
if (ecWorker != null) {
ecWorker.shutDown();
}
if(blockPoolManager != null) { if(blockPoolManager != null) {
try { try {
this.blockPoolManager.shutDownAll(bposArray); this.blockPoolManager.shutDownAll(bposArray);

11
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/erasurecode/ErasureCodingWorker.java
View File

@ -27,6 +27,8 @@
import org.slf4j.Logger; import org.slf4j.Logger;
import java.util.Collection; import java.util.Collection;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue; import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor;
@ -149,7 +151,12 @@ Configuration getConf() {
return conf; return conf;
} }
ThreadPoolExecutor getStripedReadPool() { CompletionService<Void> createReadService() {
return stripedReadPool; return new ExecutorCompletionService<>(stripedReadPool);
}
public void shutDown() {
stripedReconstructionPool.shutdown();
stripedReadPool.shutdown();
} }
} }

8
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/erasurecode/StripedReconstructor.java
View File

@ -39,8 +39,6 @@
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.util.BitSet; import java.util.BitSet;
import java.util.concurrent.CompletionService; import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
/** /**
@ -110,7 +108,7 @@ abstract class StripedReconstructor {
// position in striped internal block // position in striped internal block
private long positionInBlock; private long positionInBlock;
private StripedReader stripedReader; private StripedReader stripedReader;
private ThreadPoolExecutor stripedReadPool; private ErasureCodingWorker erasureCodingWorker;
private final CachingStrategy cachingStrategy; private final CachingStrategy cachingStrategy;
private long maxTargetLength = 0L; private long maxTargetLength = 0L;
private final BitSet liveBitSet; private final BitSet liveBitSet;
@ -122,7 +120,7 @@ abstract class StripedReconstructor {
StripedReconstructor(ErasureCodingWorker worker, StripedReconstructor(ErasureCodingWorker worker,
StripedReconstructionInfo stripedReconInfo) { StripedReconstructionInfo stripedReconInfo) {
this.stripedReadPool = worker.getStripedReadPool(); this.erasureCodingWorker = worker;
this.datanode = worker.getDatanode(); this.datanode = worker.getDatanode();
this.conf = worker.getConf(); this.conf = worker.getConf();
this.ecPolicy = stripedReconInfo.getEcPolicy(); this.ecPolicy = stripedReconInfo.getEcPolicy();
@ -225,7 +223,7 @@ CachingStrategy getCachingStrategy() {
} }
CompletionService<Void> createReadService() { CompletionService<Void> createReadService() {
return new ExecutorCompletionService<>(stripedReadPool); return erasureCodingWorker.createReadService();
} }
ExtendedBlock getBlockGroup() { ExtendedBlock getBlockGroup() {