Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-10182. Hedged read might overwrite user's buf. Contributed by zhouyingchao. 

(cherry picked from commit d8383c687c)
This commit is contained in:
Walter Su 2016-03-28 15:44:25 +08:00
parent 17dbf82f54
commit ec459e3966
1 changed files with 7 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java
View File

@ -1249,7 +1249,7 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead,
// chooseDataNode is a commitment. If no node, we go to // chooseDataNode is a commitment. If no node, we go to
// the NN to reget block locations. Only go here on first read. // the NN to reget block locations. Only go here on first read.
chosenNode = chooseDataNode(block, ignored); chosenNode = chooseDataNode(block, ignored);
bb = ByteBuffer.wrap(buf, offset, len); bb = ByteBuffer.allocate(len);
Callable<ByteBuffer> getFromDataNodeCallable = getFromOneDataNode( Callable<ByteBuffer> getFromDataNodeCallable = getFromOneDataNode(
chosenNode, block, start, end, bb, corruptedBlockMap, chosenNode, block, start, end, bb, corruptedBlockMap,
hedgedReadId++); hedgedReadId++);
@ -1260,7 +1260,9 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead,
Future<ByteBuffer> future = hedgedService.poll( Future<ByteBuffer> future = hedgedService.poll(
dfsClient.getHedgedReadTimeout(), TimeUnit.MILLISECONDS); dfsClient.getHedgedReadTimeout(), TimeUnit.MILLISECONDS);
if (future != null) { if (future != null) {
future.get(); ByteBuffer result = future.get();
System.arraycopy(result.array(), result.position(), buf, offset,
len);
return; return;
} }
if (DFSClient.LOG.isDebugEnabled()) { if (DFSClient.LOG.isDebugEnabled()) {
@ -1306,13 +1308,9 @@ implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead,
ByteBuffer result = getFirstToComplete(hedgedService, futures); ByteBuffer result = getFirstToComplete(hedgedService, futures);
// cancel the rest. // cancel the rest.
cancelAll(futures); cancelAll(futures);
if (result.array() != buf) { // compare the array pointers
dfsClient.getHedgedReadMetrics().incHedgedReadWins(); dfsClient.getHedgedReadMetrics().incHedgedReadWins();
System.arraycopy(result.array(), result.position(), buf, offset, System.arraycopy(result.array(), result.position(), buf, offset,
len); len);
} else {
dfsClient.getHedgedReadMetrics().incHedgedReadOps();
}
return; return;
} catch (InterruptedException ie) { } catch (InterruptedException ie) {
// Ignore and retry // Ignore and retry