You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/01/22 18:51:34 UTC
[flink] 01/03: [FLINK-20654][network] Improve tracing of buffers
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d087421650f48d911bdc5109b1bc8985ed27d724
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Thu Jan 21 15:45:13 2021 +0100
[FLINK-20654][network] Improve tracing of buffers
This commit provides additional contextual information that is easing debugging.
---
.../channel/ChannelStateWriteRequest.java | 9 ++-
.../channel/RecoveredChannelStateHandler.java | 8 ++-
.../flink/runtime/io/network/buffer/Buffer.java | 12 +++-
.../runtime/io/network/buffer/BufferConsumer.java | 13 ++++
.../io/network/logger/NetworkActionsLogger.java | 73 ++++++++++++++--------
.../network/partition/PipelinedSubpartition.java | 3 +
.../partition/consumer/ChannelStatePersister.java | 38 ++++++++++-
.../partition/consumer/LocalInputChannel.java | 7 +++
.../partition/consumer/RemoteInputChannel.java | 7 ++-
.../partition/consumer/SingleInputGate.java | 3 -
10 files changed, 134 insertions(+), 39 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java
index 0d347ab..e833d89 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java
@@ -59,7 +59,8 @@ interface ChannelStateWriteRequest {
long checkpointId, InputChannelInfo info, CloseableIterator<Buffer> iterator) {
return buildWriteRequest(
checkpointId,
- "writeInput",
+ "ChannelStateWriteRequest#writeInput",
+ info,
iterator,
(writer, buffer) -> writer.writeInput(info, buffer));
}
@@ -68,7 +69,8 @@ interface ChannelStateWriteRequest {
long checkpointId, ResultSubpartitionInfo info, Buffer... buffers) {
return buildWriteRequest(
checkpointId,
- "writeOutput",
+ "ChannelStateWriteRequest#writeOutput",
+ info,
ofElements(Buffer::recycleBuffer, buffers),
(writer, buffer) -> writer.writeOutput(info, buffer));
}
@@ -76,6 +78,7 @@ interface ChannelStateWriteRequest {
static ChannelStateWriteRequest buildWriteRequest(
long checkpointId,
String name,
+ Object channelInfo,
CloseableIterator<Buffer> iterator,
BiConsumerWithException<ChannelStateCheckpointWriter, Buffer, Exception>
bufferConsumer) {
@@ -85,7 +88,7 @@ interface ChannelStateWriteRequest {
writer -> {
while (iterator.hasNext()) {
Buffer buffer = iterator.next();
- NetworkActionsLogger.log(ChannelStateWriteRequest.class, name, buffer);
+ NetworkActionsLogger.tracePersist(name, buffer, channelInfo, checkpointId);
try {
checkArgument(buffer.isBuffer());
} catch (Exception e) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
index f2ccfb6..eb72ac7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
@@ -67,7 +67,8 @@ class InputChannelRecoveredStateHandler
@Override
public void recover(InputChannelInfo channelInfo, Buffer buffer) {
if (buffer.readableBytes() > 0) {
- NetworkActionsLogger.log(getClass(), "recover", buffer);
+ NetworkActionsLogger.traceRecover(
+ "InputChannelRecoveredStateHandler#recover", buffer, channelInfo);
getChannel(channelInfo).onRecoveredStateBuffer(buffer);
} else {
buffer.recycleBuffer();
@@ -118,7 +119,10 @@ class ResultSubpartitionRecoveredStateHandler
throws IOException {
bufferBuilderAndConsumer.f0.finish();
if (bufferBuilderAndConsumer.f1.isDataAvailable()) {
- NetworkActionsLogger.log(getClass(), "recover", bufferBuilderAndConsumer.f1);
+ NetworkActionsLogger.traceRecover(
+ "ResultSubpartitionRecoveredStateHandler#recover",
+ bufferBuilderAndConsumer.f1,
+ subpartitionInfo);
boolean added =
getSubpartition(subpartitionInfo)
.add(bufferBuilderAndConsumer.f1, Integer.MIN_VALUE);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
index c81505c..1dd73d6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/Buffer.java
@@ -27,6 +27,7 @@ import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBufAllocator;
import java.nio.ByteBuffer;
+import java.util.Arrays;
import static org.apache.flink.util.Preconditions.checkState;
@@ -52,7 +53,6 @@ import static org.apache.flink.util.Preconditions.checkState;
* and {@link #setSize(int)}.
*/
public interface Buffer {
-
/**
* Returns whether this buffer represents a buffer or an event.
*
@@ -227,6 +227,16 @@ public interface Buffer {
/** Sets the type of data this buffer represents. */
void setDataType(DataType dataType);
+ default String toDebugString(boolean includeHash) {
+ StringBuilder prettyString = new StringBuilder("Buffer{size=").append(getSize());
+ if (includeHash) {
+ byte[] bytes = new byte[getSize()];
+ readOnlySlice().asByteBuf().readBytes(bytes);
+ prettyString.append(", hash=").append(Arrays.hashCode(bytes));
+ }
+ return prettyString.append("}").toString();
+ }
+
/**
* Used to identify the type of data contained in the {@link Buffer} so that we can get the
* information without deserializing the serialized data.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
index e9aa8cb..191d899 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferConsumer.java
@@ -197,6 +197,19 @@ public class BufferConsumer implements Closeable {
return currentReaderPosition < writerPosition.getLatest();
}
+ public String toDebugString(boolean includeHash) {
+ Buffer buffer = null;
+ try (BufferConsumer copiedBufferConsumer = copy()) {
+ buffer = copiedBufferConsumer.build();
+ checkState(copiedBufferConsumer.isFinished());
+ return buffer.toDebugString(includeHash);
+ } finally {
+ if (buffer != null) {
+ buffer.recycleBuffer();
+ }
+ }
+ }
+
/**
* Cached reading wrapper around {@link PositionMarker}.
*
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/logger/NetworkActionsLogger.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/logger/NetworkActionsLogger.java
index 0aed4d6..4ecf18c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/logger/NetworkActionsLogger.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/logger/NetworkActionsLogger.java
@@ -18,51 +18,72 @@
package org.apache.flink.runtime.io.network.logger;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionInfo;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferConsumer;
+import org.apache.flink.runtime.io.network.partition.consumer.ChannelStatePersister;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Arrays;
-
-import static org.apache.flink.util.Preconditions.checkState;
-
-/** Utility class for logging actions that happened in the network stack for debugging purposes. */
+/**
+ * Utility class for logging actions that happened in the network stack for debugging purposes.
+ *
+ * <p>Action parameter typically includes class and method names.
+ */
public class NetworkActionsLogger {
private static final Logger LOG = LoggerFactory.getLogger(NetworkActionsLogger.class);
-
private static final boolean ENABLED = LOG.isTraceEnabled();
private static final boolean INCLUDE_HASH = true;
- public static void log(Class<?> clazz, String action, Buffer buffer) {
+ public static void traceInput(
+ String action,
+ Buffer buffer,
+ InputChannelInfo channelInfo,
+ ChannelStatePersister channelStatePersister,
+ int sequenceNumber) {
+ if (ENABLED) {
+ LOG.trace(
+ "{} {}, seq {}, {} @ {}",
+ action,
+ buffer.toDebugString(INCLUDE_HASH),
+ sequenceNumber,
+ channelStatePersister,
+ channelInfo);
+ }
+ }
+
+ public static void traceOutput(
+ String action, Buffer buffer, ResultSubpartitionInfo channelInfo) {
if (ENABLED) {
- LOG.trace("{}#{} buffer = [{}]", clazz.getSimpleName(), action, toPrettyString(buffer));
+ LOG.trace("{} {} @ {}", action, buffer.toDebugString(INCLUDE_HASH), channelInfo);
}
}
- public static void log(Class<?> clazz, String action, BufferConsumer bufferConsumer) {
+ public static void traceRecover(String action, Buffer buffer, InputChannelInfo channelInfo) {
if (ENABLED) {
- Buffer buffer = null;
- try (BufferConsumer copiedBufferConsumer = bufferConsumer.copy()) {
- buffer = copiedBufferConsumer.build();
- log(clazz, action, buffer);
- checkState(copiedBufferConsumer.isFinished());
- } finally {
- if (buffer != null) {
- buffer.recycleBuffer();
- }
- }
+ LOG.trace("{} {} @ {}", action, buffer.toDebugString(INCLUDE_HASH), channelInfo);
}
}
- private static String toPrettyString(Buffer buffer) {
- StringBuilder prettyString = new StringBuilder("size=").append(buffer.getSize());
- if (INCLUDE_HASH) {
- byte[] bytes = new byte[buffer.getSize()];
- buffer.readOnlySlice().asByteBuf().readBytes(bytes);
- prettyString.append(", hash=").append(Arrays.hashCode(bytes));
+ public static void traceRecover(
+ String action, BufferConsumer bufferConsumer, ResultSubpartitionInfo channelInfo) {
+ if (ENABLED) {
+ LOG.trace(
+ "{} {} @ {}", action, bufferConsumer.toDebugString(INCLUDE_HASH), channelInfo);
+ }
+ }
+
+ public static void tracePersist(
+ String action, Buffer buffer, Object channelInfo, long checkpointId) {
+ if (ENABLED) {
+ LOG.trace(
+ "{} {}, checkpoint {} @ {}",
+ action,
+ buffer.toDebugString(INCLUDE_HASH),
+ checkpointId,
+ channelInfo);
}
- return prettyString.toString();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index 4e05161..422c832 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
import org.apache.flink.runtime.io.network.buffer.BufferConsumer;
import org.apache.flink.runtime.io.network.buffer.BufferConsumerWithPartialRecordLength;
+import org.apache.flink.runtime.io.network.logger.NetworkActionsLogger;
import org.apache.flink.runtime.io.network.partition.consumer.EndOfChannelStateEvent;
import org.apache.flink.shaded.guava18.com.google.common.collect.Iterators;
@@ -323,6 +324,8 @@ public class PipelinedSubpartition extends ResultSubpartition
// It will be reported for reading either on flush or when the number of buffers in the
// queue
// will be 2 or more.
+ NetworkActionsLogger.traceOutput(
+ "PipelinedSubpartition#pollBuffer", buffer, subpartitionInfo);
return new BufferAndBacklog(
buffer,
getBuffersInBacklog(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/ChannelStatePersister.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/ChannelStatePersister.java
index 00be6cb..59a9181 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/ChannelStatePersister.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/ChannelStatePersister.java
@@ -25,6 +25,9 @@ import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.util.CloseableIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
@@ -36,7 +39,9 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** Helper class for persisting channel state via {@link ChannelStateWriter}. */
@NotThreadSafe
-final class ChannelStatePersister {
+public final class ChannelStatePersister {
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelStatePersister.class);
+
private final InputChannelInfo channelInfo;
private enum CheckpointStatus {
@@ -61,6 +66,7 @@ final class ChannelStatePersister {
}
protected void startPersisting(long barrierId, List<Buffer> knownBuffers) {
+ logEvent("startPersisting", barrierId);
if (checkpointStatus != CheckpointStatus.BARRIER_RECEIVED && lastSeenBarrier < barrierId) {
checkpointStatus = CheckpointStatus.BARRIER_PENDING;
lastSeenBarrier = barrierId;
@@ -75,6 +81,7 @@ final class ChannelStatePersister {
}
protected void stopPersisting(long id) {
+ logEvent("stopPersisting", id);
if (id >= lastSeenBarrier) {
checkpointStatus = CheckpointStatus.COMPLETED;
lastSeenBarrier = id;
@@ -94,15 +101,31 @@ final class ChannelStatePersister {
protected Optional<Long> checkForBarrier(Buffer buffer) throws IOException {
final AbstractEvent priorityEvent = parsePriorityEvent(buffer);
if (priorityEvent instanceof CheckpointBarrier) {
- if (((CheckpointBarrier) priorityEvent).getId() >= lastSeenBarrier) {
+ final long barrierId = ((CheckpointBarrier) priorityEvent).getId();
+ if (barrierId >= lastSeenBarrier) {
+ logEvent("found barrier", barrierId);
checkpointStatus = CheckpointStatus.BARRIER_RECEIVED;
- lastSeenBarrier = ((CheckpointBarrier) priorityEvent).getId();
+ lastSeenBarrier = barrierId;
return Optional.of(lastSeenBarrier);
+ } else {
+ logEvent("ignoring barrier", barrierId);
}
}
return Optional.empty();
}
+ private void logEvent(String event, long barrierId) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "{} {}, lastSeenBarrier = {} ({}) @ {}",
+ event,
+ barrierId,
+ lastSeenBarrier,
+ checkpointStatus,
+ channelInfo);
+ }
+ }
+
/**
* Parses the buffer as an event and returns the {@link CheckpointBarrier} if the event is
* indeed a barrier or returns null in all other cases.
@@ -124,4 +147,13 @@ final class ChannelStatePersister {
protected boolean hasBarrierReceived() {
return checkpointStatus == CheckpointStatus.BARRIER_RECEIVED;
}
+
+ @Override
+ public String toString() {
+ return "ChannelStatePersister(lastSeenBarrier="
+ + lastSeenBarrier
+ + " ("
+ + checkpointStatus
+ + ")}";
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
index 5260b8d..21f415b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
@@ -27,6 +27,7 @@ import org.apache.flink.runtime.io.network.TaskEventPublisher;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.FileRegionBuffer;
+import org.apache.flink.runtime.io.network.logger.NetworkActionsLogger;
import org.apache.flink.runtime.io.network.partition.BufferAvailabilityListener;
import org.apache.flink.runtime.io.network.partition.PartitionNotFoundException;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
@@ -258,6 +259,12 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit
} else {
channelStatePersister.maybePersist(buffer);
}
+ NetworkActionsLogger.traceInput(
+ "LocalInputChannel#getNextBuffer",
+ buffer,
+ channelInfo,
+ channelStatePersister,
+ next.getSequenceNumber());
return Optional.of(
new BufferAndAvailability(
buffer,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
index 327a8ba..3090f47 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
@@ -445,7 +445,12 @@ public class RemoteInputChannel extends InputChannel {
final boolean wasEmpty;
boolean firstPriorityEvent = false;
synchronized (receivedBuffers) {
- NetworkActionsLogger.log(getClass(), "onBuffer", buffer);
+ NetworkActionsLogger.traceInput(
+ "RemoteInputChannel#onBuffer",
+ buffer,
+ channelInfo,
+ channelStatePersister,
+ sequenceNumber);
// Similar to notifyBufferAvailable(), make sure that we never add a buffer
// after releaseAllResources() released all buffers from receivedBuffers
// (see above for details).
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
index 987882c..1345c3b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
@@ -33,7 +33,6 @@ import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferDecompressor;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.BufferProvider;
-import org.apache.flink.runtime.io.network.logger.NetworkActionsLogger;
import org.apache.flink.runtime.io.network.partition.PartitionProducerStateProvider;
import org.apache.flink.runtime.io.network.partition.PrioritizedDeque;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
@@ -653,8 +652,6 @@ public class SingleInputGate extends IndexedInputGate {
checkUnavailability();
continue;
}
- NetworkActionsLogger.log(
- getClass(), "waitAndGetNextData", bufferAndAvailabilityOpt.get().buffer());
final BufferAndAvailability bufferAndAvailability = bufferAndAvailabilityOpt.get();
if (bufferAndAvailability.moreAvailable()) {