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

HDFS-3979. For hsync, datanode should wait for the local sync to complete before sending ack. Contributed by Lars Hofhansl 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1406382 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Tsz-wo Sze 2012-11-06 22:34:29 +00:00
parent 251230a126
commit 1e7010cf38
2 changed files with 26 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
View File

@ -558,6 +558,9 @@ Release 2.0.3-alpha - Unreleased
HDFS-1331. dfs -test should work like /bin/test (Andy Isaacson via daryn) HDFS-1331. dfs -test should work like /bin/test (Andy Isaacson via daryn)
HDFS-3979. For hsync, datanode should wait for the local sync to complete
before sending ack. (Lars Hofhansl via szetszwo)
Release 2.0.2-alpha - 2012-09-07 Release 2.0.2-alpha - 2012-09-07
INCOMPATIBLE CHANGES INCOMPATIBLE CHANGES

33
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java
View File

@ -319,9 +319,6 @@ public void close() throws IOException {
* @throws IOException * @throws IOException
*/ */
void flushOrSync(boolean isSync) throws IOException { void flushOrSync(boolean isSync) throws IOException {
if (isSync && (out != null || checksumOut != null)) {
datanode.metrics.incrFsyncCount();
}
long flushTotalNanos = 0; long flushTotalNanos = 0;
if (checksumOut != null) { if (checksumOut != null) {
long flushStartNanos = System.nanoTime(); long flushStartNanos = System.nanoTime();
@ -347,6 +344,9 @@ void flushOrSync(boolean isSync) throws IOException {
} }
if (checksumOut != null || out != null) { if (checksumOut != null || out != null) {
datanode.metrics.addFlushNanos(flushTotalNanos); datanode.metrics.addFlushNanos(flushTotalNanos);
if (isSync) {
datanode.metrics.incrFsyncCount();
}
} }
} }
@ -438,8 +438,10 @@ private int receivePacket() throws IOException {
int len = header.getDataLen(); int len = header.getDataLen();
boolean syncBlock = header.getSyncBlock(); boolean syncBlock = header.getSyncBlock();
// make sure the block gets sync'ed upon close // avoid double sync'ing on close
this.syncOnClose |= syncBlock && lastPacketInBlock; if (syncBlock && lastPacketInBlock) {
this.syncOnClose = false;
}
// update received bytes // update received bytes
long firstByteInBlock = offsetInBlock; long firstByteInBlock = offsetInBlock;
@ -448,9 +450,9 @@ private int receivePacket() throws IOException {
replicaInfo.setNumBytes(offsetInBlock); replicaInfo.setNumBytes(offsetInBlock);
} }
// put in queue for pending acks // put in queue for pending acks, unless sync was requested
if (responder != null) { if (responder != null && !syncBlock) {
((PacketResponder)responder.getRunnable()).enqueue(seqno, ((PacketResponder) responder.getRunnable()).enqueue(seqno,
lastPacketInBlock, offsetInBlock); lastPacketInBlock, offsetInBlock);
} }
@ -471,8 +473,8 @@ private int receivePacket() throws IOException {
if(LOG.isDebugEnabled()) { if(LOG.isDebugEnabled()) {
LOG.debug("Receiving an empty packet or the end of the block " + block); LOG.debug("Receiving an empty packet or the end of the block " + block);
} }
// flush unless close() would flush anyway // sync block if requested
if (syncBlock && !lastPacketInBlock) { if (syncBlock) {
flushOrSync(true); flushOrSync(true);
} }
} else { } else {
@ -563,8 +565,8 @@ private int receivePacket() throws IOException {
checksumBuf.arrayOffset() + checksumBuf.position(), checksumBuf.arrayOffset() + checksumBuf.position(),
checksumLen); checksumLen);
} }
/// flush entire packet, sync unless close() will sync /// flush entire packet, sync if requested
flushOrSync(syncBlock && !lastPacketInBlock); flushOrSync(syncBlock);
replicaInfo.setLastChecksumAndDataLen( replicaInfo.setLastChecksumAndDataLen(
offsetInBlock, lastChunkChecksum offsetInBlock, lastChunkChecksum
@ -580,6 +582,13 @@ private int receivePacket() throws IOException {
} }
} }
// if sync was requested, put in queue for pending acks here
// (after the fsync finished)
if (responder != null && syncBlock) {
((PacketResponder) responder.getRunnable()).enqueue(seqno,
lastPacketInBlock, offsetInBlock);
}
if (throttler != null) { // throttle I/O if (throttler != null) { // throttle I/O
throttler.throttle(len); throttler.throttle(len);
} }