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

HDFS-17273. Change the way of computing some local variables duration for better debugging (#6321). Contributed by farmmamba.

Browse Source 
Reviewed-by: Haiyang Hu <haiyang.hu@shopee.com>
Reviewed-by: Tao Li <tomscut@apache.org>
Signed-off-by: He Xiaoqiao <hexiaoqiao@apache.org>
This commit is contained in:
hfutatzhanghb 2023-12-24 16:57:04 +08:00 committed by GitHub
parent 77edca8f0a
commit 415e9bdfbd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 13 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DataStreamer.java
View File

@ -787,7 +787,7 @@ public void run() {
scope = null; scope = null;
dataQueue.removeFirst(); dataQueue.removeFirst();
ackQueue.addLast(one); ackQueue.addLast(one);
packetSendTime.put(one.getSeqno(), Time.monotonicNow()); packetSendTime.put(one.getSeqno(), Time.monotonicNowNanos());
dataQueue.notifyAll(); dataQueue.notifyAll();
} }
} }
@ -953,7 +953,7 @@ void waitForAckedSeqno(long seqno) throws IOException {
dnodes = nodes != null ? nodes.length : 3; dnodes = nodes != null ? nodes.length : 3;
} }
int writeTimeout = dfsClient.getDatanodeWriteTimeout(dnodes); int writeTimeout = dfsClient.getDatanodeWriteTimeout(dnodes);
long begin = Time.monotonicNow(); long begin = Time.monotonicNowNanos();
try { try {
synchronized (dataQueue) { synchronized (dataQueue) {
while (!streamerClosed) { while (!streamerClosed) {
@ -963,14 +963,14 @@ void waitForAckedSeqno(long seqno) throws IOException {
} }
try { try {
dataQueue.wait(1000); // when we receive an ack, we notify on dataQueue.wait(1000); // when we receive an ack, we notify on
long duration = Time.monotonicNow() - begin; long duration = Time.monotonicNowNanos() - begin;
if (duration > writeTimeout) { if (TimeUnit.NANOSECONDS.toMillis(duration) > writeTimeout) {
LOG.error("No ack received, took {}ms (threshold={}ms). " LOG.error("No ack received, took {}ms (threshold={}ms). "
+ "File being written: {}, block: {}, " + "File being written: {}, block: {}, "
+ "Write pipeline datanodes: {}.", + "Write pipeline datanodes: {}.",
duration, writeTimeout, src, block, nodes); TimeUnit.NANOSECONDS.toMillis(duration), writeTimeout, src, block, nodes);
throw new InterruptedIOException("No ack received after " + throw new InterruptedIOException("No ack received after " +
duration / 1000 + "s and a timeout of " + TimeUnit.NANOSECONDS.toSeconds(duration) + "s and a timeout of " +
writeTimeout / 1000 + "s"); writeTimeout / 1000 + "s");
} }
// dataQueue // dataQueue
@ -984,11 +984,12 @@ void waitForAckedSeqno(long seqno) throws IOException {
} catch (ClosedChannelException cce) { } catch (ClosedChannelException cce) {
LOG.debug("Closed channel exception", cce); LOG.debug("Closed channel exception", cce);
} }
long duration = Time.monotonicNow() - begin; long duration = Time.monotonicNowNanos() - begin;
if (duration > dfsclientSlowLogThresholdMs) { if (TimeUnit.NANOSECONDS.toMillis(duration) > dfsclientSlowLogThresholdMs) {
LOG.warn("Slow waitForAckedSeqno took {}ms (threshold={}ms). File being" LOG.warn("Slow waitForAckedSeqno took {}ms (threshold={}ms). File being"
+ " written: {}, block: {}, Write pipeline datanodes: {}.", + " written: {}, block: {}, Write pipeline datanodes: {}.",
duration, dfsclientSlowLogThresholdMs, src, block, nodes); TimeUnit.NANOSECONDS.toMillis(duration), dfsclientSlowLogThresholdMs,
src, block, nodes);
} }
} }
} }
@ -1179,10 +1180,10 @@ public void run() {
if (ack.getSeqno() != DFSPacket.HEART_BEAT_SEQNO) { if (ack.getSeqno() != DFSPacket.HEART_BEAT_SEQNO) {
Long begin = packetSendTime.get(ack.getSeqno()); Long begin = packetSendTime.get(ack.getSeqno());
if (begin != null) { if (begin != null) {
long duration = Time.monotonicNow() - begin; long duration = Time.monotonicNowNanos() - begin;
if (duration > dfsclientSlowLogThresholdMs) { if (TimeUnit.NANOSECONDS.toMillis(duration) > dfsclientSlowLogThresholdMs) {
LOG.info("Slow ReadProcessor read fields for block " + block LOG.info("Slow ReadProcessor read fields for block " + block
+ " took " + duration + "ms (threshold=" + " took " + TimeUnit.NANOSECONDS.toMillis(duration) + "ms (threshold="
+ dfsclientSlowLogThresholdMs + "ms); ack: " + ack + dfsclientSlowLogThresholdMs + "ms); ack: " + ack
+ ", targets: " + Arrays.asList(targets)); + ", targets: " + Arrays.asList(targets));
} }