|
|
@@ -114,14 +114,39 @@ public class Dispatcher {
|
|
|
|
|
|
private NetworkTopology cluster;
|
|
|
|
|
|
- private final ExecutorService moveExecutor;
|
|
|
private final ExecutorService dispatchExecutor;
|
|
|
|
|
|
+ private final Allocator moverThreadAllocator;
|
|
|
+
|
|
|
/** The maximum number of concurrent blocks moves at a datanode */
|
|
|
private final int maxConcurrentMovesPerNode;
|
|
|
|
|
|
private final int ioFileBufferSize;
|
|
|
|
|
|
+ static class Allocator {
|
|
|
+ private final int max;
|
|
|
+ private int count = 0;
|
|
|
+
|
|
|
+ Allocator(int max) {
|
|
|
+ this.max = max;
|
|
|
+ }
|
|
|
+
|
|
|
+ synchronized int allocate(int n) {
|
|
|
+ final int remaining = max - count;
|
|
|
+ if (remaining <= 0) {
|
|
|
+ return 0;
|
|
|
+ } else {
|
|
|
+ final int allocated = remaining < n? remaining: n;
|
|
|
+ count += allocated;
|
|
|
+ return allocated;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ synchronized void reset() {
|
|
|
+ count = 0;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
private static class GlobalBlockMap {
|
|
|
private final Map<Block, DBlock> map = new HashMap<Block, DBlock>();
|
|
|
|
|
|
@@ -287,9 +312,7 @@ public class Dispatcher {
|
|
|
|
|
|
/** Dispatch the move to the proxy source & wait for the response. */
|
|
|
private void dispatch() {
|
|
|
- if (LOG.isDebugEnabled()) {
|
|
|
- LOG.debug("Start moving " + this);
|
|
|
- }
|
|
|
+ LOG.info("Start moving " + this);
|
|
|
|
|
|
Socket sock = new Socket();
|
|
|
DataOutputStream out = null;
|
|
|
@@ -504,7 +527,7 @@ public class Dispatcher {
|
|
|
private final List<PendingMove> pendings;
|
|
|
private volatile boolean hasFailure = false;
|
|
|
private volatile boolean hasSuccess = false;
|
|
|
- private final int maxConcurrentMoves;
|
|
|
+ private ExecutorService moveExecutor;
|
|
|
|
|
|
@Override
|
|
|
public String toString() {
|
|
|
@@ -513,7 +536,6 @@ public class Dispatcher {
|
|
|
|
|
|
private DDatanode(DatanodeInfo datanode, int maxConcurrentMoves) {
|
|
|
this.datanode = datanode;
|
|
|
- this.maxConcurrentMoves = maxConcurrentMoves;
|
|
|
this.pendings = new ArrayList<PendingMove>(maxConcurrentMoves);
|
|
|
}
|
|
|
|
|
|
@@ -521,6 +543,21 @@ public class Dispatcher {
|
|
|
return datanode;
|
|
|
}
|
|
|
|
|
|
+ synchronized ExecutorService initMoveExecutor(int poolSize) {
|
|
|
+ return moveExecutor = Executors.newFixedThreadPool(poolSize);
|
|
|
+ }
|
|
|
+
|
|
|
+ synchronized ExecutorService getMoveExecutor() {
|
|
|
+ return moveExecutor;
|
|
|
+ }
|
|
|
+
|
|
|
+ synchronized void shutdownMoveExecutor() {
|
|
|
+ if (moveExecutor != null) {
|
|
|
+ moveExecutor.shutdown();
|
|
|
+ moveExecutor = null;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
private static <G extends StorageGroup> void put(StorageType storageType,
|
|
|
G g, EnumMap<StorageType, G> map) {
|
|
|
final StorageGroup existing = map.put(storageType, g);
|
|
|
@@ -541,6 +578,7 @@ public class Dispatcher {
|
|
|
|
|
|
synchronized private void activateDelay(long delta) {
|
|
|
delayUntil = Time.monotonicNow() + delta;
|
|
|
+ LOG.info(this + " activateDelay " + delta/1000.0 + " seconds");
|
|
|
}
|
|
|
|
|
|
synchronized private boolean isDelayActive() {
|
|
|
@@ -551,11 +589,6 @@ public class Dispatcher {
|
|
|
return true;
|
|
|
}
|
|
|
|
|
|
- /** Check if the node can schedule more blocks to move */
|
|
|
- synchronized boolean isPendingQNotFull() {
|
|
|
- return pendings.size() < maxConcurrentMoves;
|
|
|
- }
|
|
|
-
|
|
|
/** Check if all the dispatched moves are done */
|
|
|
synchronized boolean isPendingQEmpty() {
|
|
|
return pendings.isEmpty();
|
|
|
@@ -563,7 +596,7 @@ public class Dispatcher {
|
|
|
|
|
|
/** Add a scheduled block move to the node */
|
|
|
synchronized boolean addPendingBlock(PendingMove pendingBlock) {
|
|
|
- if (!isDelayActive() && isPendingQNotFull()) {
|
|
|
+ if (!isDelayActive()) {
|
|
|
return pendings.add(pendingBlock);
|
|
|
}
|
|
|
return false;
|
|
|
@@ -621,6 +654,11 @@ public class Dispatcher {
|
|
|
private long getBlockList() throws IOException {
|
|
|
final long size = Math.min(MAX_BLOCKS_SIZE_TO_FETCH, blocksToReceive);
|
|
|
final BlocksWithLocations newBlocks = nnc.getBlocks(getDatanodeInfo(), size);
|
|
|
+ if (LOG.isTraceEnabled()) {
|
|
|
+ LOG.trace("getBlocks(" + getDatanodeInfo() + ", "
|
|
|
+ + StringUtils.TraditionalBinaryPrefix.long2String(size, "B", 2)
|
|
|
+ + ") returns " + newBlocks.getBlocks().length + " blocks.");
|
|
|
+ }
|
|
|
|
|
|
long bytesReceived = 0;
|
|
|
for (BlockWithLocations blk : newBlocks.getBlocks()) {
|
|
|
@@ -642,7 +680,9 @@ public class Dispatcher {
|
|
|
}
|
|
|
}
|
|
|
if (!srcBlocks.contains(block) && isGoodBlockCandidate(block)) {
|
|
|
- // filter bad candidates
|
|
|
+ if (LOG.isTraceEnabled()) {
|
|
|
+ LOG.trace("Add " + block + " to " + this);
|
|
|
+ }
|
|
|
srcBlocks.add(block);
|
|
|
}
|
|
|
}
|
|
|
@@ -710,11 +750,9 @@ public class Dispatcher {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- private static final int SOURCE_BLOCKS_MIN_SIZE = 5;
|
|
|
-
|
|
|
/** @return if should fetch more blocks from namenode */
|
|
|
private boolean shouldFetchMoreBlocks() {
|
|
|
- return srcBlocks.size() < SOURCE_BLOCKS_MIN_SIZE && blocksToReceive > 0;
|
|
|
+ return blocksToReceive > 0;
|
|
|
}
|
|
|
|
|
|
private static final long MAX_ITERATION_TIME = 20 * 60 * 1000L; // 20 mins
|
|
|
@@ -734,6 +772,11 @@ public class Dispatcher {
|
|
|
int noPendingMoveIteration = 0;
|
|
|
while (!isTimeUp && getScheduledSize() > 0
|
|
|
&& (!srcBlocks.isEmpty() || blocksToReceive > 0)) {
|
|
|
+ if (LOG.isTraceEnabled()) {
|
|
|
+ LOG.trace(this + " blocksToReceive=" + blocksToReceive
|
|
|
+ + ", scheduledSize=" + getScheduledSize()
|
|
|
+ + ", srcBlocks#=" + srcBlocks.size());
|
|
|
+ }
|
|
|
final PendingMove p = chooseNextMove();
|
|
|
if (p != null) {
|
|
|
// Reset no pending move counter
|
|
|
@@ -761,12 +804,16 @@ public class Dispatcher {
|
|
|
// in case no blocks can be moved for source node's task,
|
|
|
// jump out of while-loop after 5 iterations.
|
|
|
if (noPendingMoveIteration >= MAX_NO_PENDING_MOVE_ITERATIONS) {
|
|
|
+ LOG.info("Failed to find a pending move " + noPendingMoveIteration
|
|
|
+ + " times. Skipping " + this);
|
|
|
resetScheduledSize();
|
|
|
}
|
|
|
}
|
|
|
|
|
|
// check if time is up or not
|
|
|
if (Time.monotonicNow() - startTime > MAX_ITERATION_TIME) {
|
|
|
+ LOG.info("Time up (max time=" + MAX_ITERATION_TIME/1000
|
|
|
+ + " seconds). Skipping " + this);
|
|
|
isTimeUp = true;
|
|
|
continue;
|
|
|
}
|
|
|
@@ -803,9 +850,9 @@ public class Dispatcher {
|
|
|
|
|
|
this.cluster = NetworkTopology.getInstance(conf);
|
|
|
|
|
|
- this.moveExecutor = Executors.newFixedThreadPool(moverThreads);
|
|
|
this.dispatchExecutor = dispatcherThreads == 0? null
|
|
|
: Executors.newFixedThreadPool(dispatcherThreads);
|
|
|
+ this.moverThreadAllocator = new Allocator(moverThreads);
|
|
|
this.maxConcurrentMovesPerNode = maxConcurrentMovesPerNode;
|
|
|
|
|
|
this.saslClient = new SaslDataTransferClient(conf,
|
|
|
@@ -890,8 +937,22 @@ public class Dispatcher {
|
|
|
return new DDatanode(datanode, maxConcurrentMovesPerNode);
|
|
|
}
|
|
|
|
|
|
+
|
|
|
public void executePendingMove(final PendingMove p) {
|
|
|
// move the block
|
|
|
+ final DDatanode targetDn = p.target.getDDatanode();
|
|
|
+ ExecutorService moveExecutor = targetDn.getMoveExecutor();
|
|
|
+ if (moveExecutor == null) {
|
|
|
+ final int nThreads = moverThreadAllocator.allocate(maxConcurrentMovesPerNode);
|
|
|
+ if (nThreads > 0) {
|
|
|
+ moveExecutor = targetDn.initMoveExecutor(nThreads);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (moveExecutor == null) {
|
|
|
+ LOG.warn("No mover threads available: skip moving " + p);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
moveExecutor.execute(new Runnable() {
|
|
|
@Override
|
|
|
public void run() {
|
|
|
@@ -1083,6 +1144,11 @@ public class Dispatcher {
|
|
|
cluster = NetworkTopology.getInstance(conf);
|
|
|
storageGroupMap.clear();
|
|
|
sources.clear();
|
|
|
+
|
|
|
+ moverThreadAllocator.reset();
|
|
|
+ for(StorageGroup t : targets) {
|
|
|
+ t.getDDatanode().shutdownMoveExecutor();
|
|
|
+ }
|
|
|
targets.clear();
|
|
|
globalBlocks.removeAllButRetain(movedBlocks);
|
|
|
movedBlocks.cleanup();
|
|
|
@@ -1104,7 +1170,6 @@ public class Dispatcher {
|
|
|
if (dispatchExecutor != null) {
|
|
|
dispatchExecutor.shutdownNow();
|
|
|
}
|
|
|
- moveExecutor.shutdownNow();
|
|
|
}
|
|
|
|
|
|
static class Util {
|
Tsz-Wo Nicholas Sze