You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/07/24 08:22:11 UTC
[flink] branch master updated (c8dfb2a1c68 -> eab99b6de82)
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from c8dfb2a1c68 [FLINK-32430][table] Support configuring CatalogStore using configuration
new 51b0df65886 [FLINK-32640][Network] Measure backpressure time for legacy hybrid shuffle.
new eab99b6de82 [FLINK-32640][Network] Measure backpressure time for tiered storage hybrid shuffle.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../partition/hybrid/HsMemoryDataManager.java | 19 +++++++++++++-
.../network/partition/hybrid/HsOutputMetrics.java | 11 +++++++-
.../partition/hybrid/HsResultPartition.java | 6 ++++-
.../tiered/shuffle/TieredResultPartition.java | 1 +
.../tiered/storage/TieredStorageMemoryManager.java | 8 ++++++
.../storage/TieredStorageMemoryManagerImpl.java | 30 ++++++++++++++++++----
.../partition/hybrid/HybridShuffleTestUtils.java | 3 ++-
.../tiered/TestingTieredStorageMemoryManager.java | 13 ++++++++++
8 files changed, 82 insertions(+), 9 deletions(-)
[flink] 02/02: [FLINK-32640][Network] Measure backpressure time for tiered storage hybrid shuffle.
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit eab99b6de820352490cce580fee2e7f31ef16617
Author: Weijie Guo <re...@163.com>
AuthorDate: Fri Jul 21 11:43:16 2023 +0800
[FLINK-32640][Network] Measure backpressure time for tiered storage hybrid shuffle.
---
.../tiered/shuffle/TieredResultPartition.java | 1 +
.../tiered/storage/TieredStorageMemoryManager.java | 8 ++++++
.../storage/TieredStorageMemoryManagerImpl.java | 30 ++++++++++++++++++----
.../tiered/TestingTieredStorageMemoryManager.java | 13 ++++++++++
4 files changed, 47 insertions(+), 5 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartition.java
index f5601fbd0ae..d3aae631620 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartition.java
@@ -122,6 +122,7 @@ public class TieredResultPartition extends ResultPartition {
@Override
public void setMetricGroup(TaskIOMetricGroup metrics) {
super.setMetricGroup(metrics);
+ storageMemoryManager.setMetricGroup(metrics);
tieredStorageProducerClient.setMetricStatisticsUpdater(
this::updateProducerMetricStatistics);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMemoryManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMemoryManager.java
index f3aa2a87eb5..6b44eba76ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMemoryManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMemoryManager.java
@@ -22,6 +22,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.BufferPool;
import org.apache.flink.runtime.io.network.buffer.LocalBufferPool;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import java.util.List;
@@ -54,6 +55,13 @@ public interface TieredStorageMemoryManager {
*/
void setup(BufferPool bufferPool, List<TieredStorageMemorySpec> storageMemorySpecs);
+ /**
+ * Set the {@link TaskIOMetricGroup} for this memory manager.
+ *
+ * @param metricGroup the metric group to set
+ */
+ void setMetricGroup(TaskIOMetricGroup metricGroup);
+
/**
* Register a listener to listen the buffer reclaim request from the {@link
* TieredStorageMemoryManager}.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMemoryManagerImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMemoryManagerImpl.java
index 7eadf0b0908..f11a413c194 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMemoryManagerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMemoryManagerImpl.java
@@ -23,6 +23,8 @@ 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.BufferPool;
import org.apache.flink.runtime.io.network.buffer.LocalBufferPool;
+import org.apache.flink.runtime.metrics.TimerGauge;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FatalExitExceptionHandler;
@@ -86,6 +88,12 @@ public class TieredStorageMemoryManagerImpl implements TieredStorageMemoryManage
*/
private final Map<Object, Integer> numOwnerRequestedBuffers;
+ /**
+ * Time gauge to measure that hard backpressure time. Pre-create it to avoid checkNotNull in
+ * hot-path for performance purpose.
+ */
+ private TimerGauge hardBackpressureTimerGauge = new TimerGauge();
+
/**
* This is for triggering buffer reclaiming while blocked on requesting new buffers.
*
@@ -145,6 +153,12 @@ public class TieredStorageMemoryManagerImpl implements TieredStorageMemoryManage
this.isInitialized = true;
}
+ @Override
+ public void setMetricGroup(TaskIOMetricGroup metricGroup) {
+ this.hardBackpressureTimerGauge =
+ checkNotNull(metricGroup.getHardBackPressuredTimePerSecond());
+ }
+
@Override
public void listenBufferReclaimRequest(Runnable onBufferReclaimRequest) {
bufferReclaimRequestListeners.add(onBufferReclaimRequest);
@@ -159,12 +173,18 @@ public class TieredStorageMemoryManagerImpl implements TieredStorageMemoryManage
CompletableFuture<Void> requestBufferFuture = new CompletableFuture<>();
scheduleCheckRequestBufferFuture(
requestBufferFuture, INITIAL_REQUEST_BUFFER_TIMEOUT_FOR_RECLAIMING_MS);
- MemorySegment memorySegment = null;
- try {
- memorySegment = bufferPool.requestMemorySegmentBlocking();
- } catch (InterruptedException e) {
- ExceptionUtils.rethrow(e);
+ MemorySegment memorySegment = bufferPool.requestMemorySegment();
+
+ if (memorySegment == null) {
+ try {
+ hardBackpressureTimerGauge.markStart();
+ memorySegment = bufferPool.requestMemorySegmentBlocking();
+ hardBackpressureTimerGauge.markEnd();
+ } catch (InterruptedException e) {
+ ExceptionUtils.rethrow(e);
+ }
}
+
requestBufferFuture.complete(null);
incNumRequestedBuffer(owner);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTieredStorageMemoryManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTieredStorageMemoryManager.java
index fbab5c87541..c99013b6a1e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTieredStorageMemoryManager.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTieredStorageMemoryManager.java
@@ -23,6 +23,7 @@ import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageMemoryManager;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageMemorySpec;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.util.function.TriConsumer;
import java.util.List;
@@ -35,6 +36,8 @@ public class TestingTieredStorageMemoryManager implements TieredStorageMemoryMan
private final BiConsumer<BufferPool, List<TieredStorageMemorySpec>> setupConsumer;
+ private final Consumer<TaskIOMetricGroup> setMetricGroupConsumer;
+
private final Consumer<Runnable> listenBufferReclaimRequestConsumer;
private final Function<Object, BufferBuilder> requestBufferBlockingFunction;
@@ -49,6 +52,7 @@ public class TestingTieredStorageMemoryManager implements TieredStorageMemoryMan
private TestingTieredStorageMemoryManager(
BiConsumer<BufferPool, List<TieredStorageMemorySpec>> setupConsumer,
+ Consumer<TaskIOMetricGroup> setMetricGroupConsumer,
Consumer<Runnable> listenBufferReclaimRequestConsumer,
Function<Object, BufferBuilder> requestBufferBlockingFunction,
Function<Object, Integer> getMaxNonReclaimableBuffersFunction,
@@ -56,6 +60,7 @@ public class TestingTieredStorageMemoryManager implements TieredStorageMemoryMan
TriConsumer<Object, Object, Buffer> transferBufferOwnershipConsumer,
Runnable releaseRunnable) {
this.setupConsumer = setupConsumer;
+ this.setMetricGroupConsumer = setMetricGroupConsumer;
this.listenBufferReclaimRequestConsumer = listenBufferReclaimRequestConsumer;
this.requestBufferBlockingFunction = requestBufferBlockingFunction;
this.getMaxNonReclaimableBuffersFunction = getMaxNonReclaimableBuffersFunction;
@@ -69,6 +74,11 @@ public class TestingTieredStorageMemoryManager implements TieredStorageMemoryMan
setupConsumer.accept(bufferPool, storageMemorySpecs);
}
+ @Override
+ public void setMetricGroup(TaskIOMetricGroup metricGroup) {
+ setMetricGroupConsumer.accept(metricGroup);
+ }
+
@Override
public void listenBufferReclaimRequest(Runnable onBufferReclaimRequest) {
listenBufferReclaimRequestConsumer.accept(onBufferReclaimRequest);
@@ -105,6 +115,8 @@ public class TestingTieredStorageMemoryManager implements TieredStorageMemoryMan
private BiConsumer<BufferPool, List<TieredStorageMemorySpec>> setupConsumer =
(bufferPool, tieredStorageMemorySpecs) -> {};
+ private Consumer<TaskIOMetricGroup> setMetricGroupConsumer = (ignore) -> {};
+
private Consumer<Runnable> listenBufferReclaimRequestConsumer = runnable -> {};
private Function<Object, BufferBuilder> requestBufferBlockingFunction = owner -> null;
@@ -165,6 +177,7 @@ public class TestingTieredStorageMemoryManager implements TieredStorageMemoryMan
public TestingTieredStorageMemoryManager build() {
return new TestingTieredStorageMemoryManager(
setupConsumer,
+ setMetricGroupConsumer,
listenBufferReclaimRequestConsumer,
requestBufferBlockingFunction,
getMaxNonReclaimableBuffersFunction,
[flink] 01/02: [FLINK-32640][Network] Measure backpressure time for legacy hybrid shuffle.
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 51b0df658864fc4caa0607fe82ce26f90c1a67ee
Author: Weijie Guo <re...@163.com>
AuthorDate: Tue Apr 25 15:38:11 2023 +0800
[FLINK-32640][Network] Measure backpressure time for legacy hybrid shuffle.
---
.../network/partition/hybrid/HsMemoryDataManager.java | 19 ++++++++++++++++++-
.../io/network/partition/hybrid/HsOutputMetrics.java | 11 ++++++++++-
.../network/partition/hybrid/HsResultPartition.java | 6 +++++-
.../partition/hybrid/HybridShuffleTestUtils.java | 3 ++-
4 files changed, 35 insertions(+), 4 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsMemoryDataManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsMemoryDataManager.java
index 387114dbf81..ac374ea27c9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsMemoryDataManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsMemoryDataManager.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.io.network.buffer.BufferBuilder;
import org.apache.flink.runtime.io.network.buffer.BufferCompressor;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.partition.hybrid.HsSpillingStrategy.Decision;
+import org.apache.flink.runtime.metrics.TimerGauge;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.concurrent.FutureUtils;
@@ -92,6 +93,12 @@ public class HsMemoryDataManager implements HsSpillingInfoProvider, HsMemoryData
private final AtomicInteger poolSize;
+ /**
+ * If task thread blocked on request buffer from buffer pool, this metric should be updated.
+ * Pre-create it to avoid checkNotNull in hot-path for performance purpose.
+ */
+ private TimerGauge hardBackPressuredTimePerSecond = new TimerGauge();
+
public HsMemoryDataManager(
int numSubpartitions,
int bufferSize,
@@ -200,6 +207,7 @@ public class HsMemoryDataManager implements HsSpillingInfoProvider, HsMemoryData
for (int i = 0; i < numSubpartitions; i++) {
getSubpartitionMemoryDataManager(i).setOutputMetrics(metrics);
}
+ this.hardBackPressuredTimePerSecond = metrics.getHardBackpressureTimerGauge();
}
// ------------------------------------
@@ -263,7 +271,16 @@ public class HsMemoryDataManager implements HsSpillingInfoProvider, HsMemoryData
@Override
public BufferBuilder requestBufferFromPool() throws InterruptedException {
- MemorySegment segment = bufferPool.requestMemorySegmentBlocking();
+ MemorySegment segment = bufferPool.requestMemorySegment();
+
+ if (segment == null) {
+ // only when the buffer is not acquired immediately, it is requested in blocking mode,
+ // which will make the calculation of backpressure more accurate.
+ hardBackPressuredTimePerSecond.markStart();
+ segment = bufferPool.requestMemorySegmentBlocking();
+ hardBackPressuredTimePerSecond.markEnd();
+ }
+
Optional<Decision> decisionOpt =
spillStrategy.onMemoryUsageChanged(
numRequestedBuffers.incrementAndGet(), getPoolSize());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsOutputMetrics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsOutputMetrics.java
index 6ffb1f03d01..ecedd1d2ebd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsOutputMetrics.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsOutputMetrics.java
@@ -19,15 +19,20 @@
package org.apache.flink.runtime.io.network.partition.hybrid;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.runtime.metrics.TimerGauge;
/** All metrics that {@link HsResultPartition} needs to count, except numBytesProduced. */
public class HsOutputMetrics {
private final Counter numBytesOut;
private final Counter numBuffersOut;
- public HsOutputMetrics(Counter numBytesOut, Counter numBuffersOut) {
+ private final TimerGauge hardBackpressureTimerGauge;
+
+ public HsOutputMetrics(
+ Counter numBytesOut, Counter numBuffersOut, TimerGauge hardBackpressureTimerGauge) {
this.numBytesOut = numBytesOut;
this.numBuffersOut = numBuffersOut;
+ this.hardBackpressureTimerGauge = hardBackpressureTimerGauge;
}
public Counter getNumBytesOut() {
@@ -37,4 +42,8 @@ public class HsOutputMetrics {
public Counter getNumBuffersOut() {
return numBuffersOut;
}
+
+ public TimerGauge getHardBackpressureTimerGauge() {
+ return hardBackpressureTimerGauge;
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartition.java
index c0298325caf..e0dca965666 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsResultPartition.java
@@ -154,7 +154,11 @@ public class HsResultPartition extends ResultPartition {
public void setMetricGroup(TaskIOMetricGroup metrics) {
super.setMetricGroup(metrics);
checkNotNull(memoryDataManager)
- .setOutputMetrics(new HsOutputMetrics(numBytesOut, numBuffersOut));
+ .setOutputMetrics(
+ new HsOutputMetrics(
+ numBytesOut,
+ numBuffersOut,
+ metrics.getHardBackPressuredTimePerSecond()));
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HybridShuffleTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HybridShuffleTestUtils.java
index 5b0629bd1a5..d9abab0b139 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HybridShuffleTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HybridShuffleTestUtils.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.io.network.partition.hybrid.HsFileDataIndexImpl.
import org.apache.flink.runtime.io.network.partition.hybrid.index.FileDataIndexRegionHelper;
import org.apache.flink.runtime.io.network.partition.hybrid.index.TestingFileDataIndexRegion;
import org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.ProducerMergedPartitionFileIndex;
+import org.apache.flink.runtime.metrics.TimerGauge;
import java.util.ArrayDeque;
import java.util.ArrayList;
@@ -74,7 +75,7 @@ public class HybridShuffleTestUtils {
}
public static HsOutputMetrics createTestingOutputMetrics() {
- return new HsOutputMetrics(new TestCounter(), new TestCounter());
+ return new HsOutputMetrics(new TestCounter(), new TestCounter(), new TimerGauge());
}
public static TestingFileDataIndexRegion createSingleTestRegion(