You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "WencongLiu (via GitHub)" <gi...@apache.org> on 2023/06/29 08:25:18 UTC
[GitHub] [flink] WencongLiu commented on a diff in pull request #22807: [FLINK-31645][network] Introduce the PartitionFileReader and DiskIOSchduler
WencongLiu commented on code in PR #22807:
URL: https://github.com/apache/flink/pull/22807#discussion_r1246296350
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileReader.java:
##########
@@ -36,9 +54,42 @@
*/
public class ProducerMergedPartitionFileReader implements PartitionFileReader {
+ /**
+ * Max number of region caches.
+ *
+ * <p>This number is utilized to constrain the total number of cached regions in the event of a
+ * region leak within the implementation. Corresponding tests have been added to guarantee
+ * correctness. As the future implementation approaches, the cache storage strategy will be
+ * established using the LRU algorithm with a justifiable cache size.
+ */
+ private static final int MAX_REGION_CACHE = 10000;
Review Comment:
Fixed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileReader.java:
##########
@@ -36,9 +54,42 @@
*/
public class ProducerMergedPartitionFileReader implements PartitionFileReader {
+ /**
+ * Max number of region caches.
+ *
+ * <p>This number is utilized to constrain the total number of cached regions in the event of a
+ * region leak within the implementation. Corresponding tests have been added to guarantee
+ * correctness. As the future implementation approaches, the cache storage strategy will be
+ * established using the LRU algorithm with a justifiable cache size.
+ */
+ private static final int MAX_REGION_CACHE = 10000;
+
+ /**
+ * Region caches stored in map.
+ *
+ * <p>The key of the cache is formed by combining the {@link TieredStorageSubpartitionId} and
+ * buffer index. The value denotes the region cache for the corresponding subpartition and
+ * buffer index. Each region cache comprises the last consumed {@link Region}, the next buffer
+ * index to consume within the region, and the file offset of the next buffer index.
+ */
+ private final Map<Tuple2<TieredStorageSubpartitionId, Integer>, RegionCache> regionCaches;
Review Comment:
Fixed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileReader.java:
##########
@@ -60,9 +141,122 @@ public long getPriority(
TieredStorageSubpartitionId subpartitionId,
int segmentId,
int bufferIndex) {
- return 0;
+ Tuple2<TieredStorageSubpartitionId, Integer> cacheKey =
+ Tuple2.of(subpartitionId, bufferIndex);
+ RegionCache regionCache = regionCaches.get(cacheKey);
+ return regionCache == null ? Long.MAX_VALUE : regionCache.getFileOffset();
}
@Override
- public void release() {}
+ public void release() {
+ if (fileChannel != null) {
+ try {
+ fileChannel.close();
+ } catch (IOException e) {
+ ExceptionUtils.rethrow(e, "Failed to close file channel.");
+ }
+ }
+ IOUtils.deleteFileQuietly(dataFilePath);
+ }
+
+ /**
+ * Initialize the file channel in a lazy manner, which can reduce usage of the file descriptor
+ * resource.
+ *
+ * @param partitionId denotes the partition id.
+ * @throws IOException is thrown in the event of an error.
+ */
+ private void lazyInitializeFileChannel(TieredStoragePartitionId partitionId)
+ throws IOException {
+ if (fileChannel == null) {
+ try {
+ fileChannel = FileChannel.open(dataFilePath, StandardOpenOption.READ);
+ } catch (FileNotFoundException e) {
+ throw new PartitionNotFoundException(
+ TieredStorageIdMappingUtils.convertId(partitionId));
+ }
+ }
+ }
+
+ /**
+ * Try to get the region cache associated with the given cache key.
+ *
+ * <p>If the relevant region cache exists within the region caches, it will be returned and
+ * subsequently removed. However, if the region cache does not exist, a new region cache will be
+ * created using the data index and returned.
+ *
+ * @param cacheKey denotes the key of the desired region cache.
+ * @return returns the relevant region cache if it exists, otherwise a null value.
+ * @throws IOException is thrown in the event of an error.
+ */
+ private RegionCache tryGetRegionCache(Tuple2<TieredStorageSubpartitionId, Integer> cacheKey)
Review Comment:
Fixed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileReader.java:
##########
@@ -60,9 +141,122 @@ public long getPriority(
TieredStorageSubpartitionId subpartitionId,
int segmentId,
int bufferIndex) {
- return 0;
+ Tuple2<TieredStorageSubpartitionId, Integer> cacheKey =
+ Tuple2.of(subpartitionId, bufferIndex);
+ RegionCache regionCache = regionCaches.get(cacheKey);
+ return regionCache == null ? Long.MAX_VALUE : regionCache.getFileOffset();
}
@Override
- public void release() {}
+ public void release() {
+ if (fileChannel != null) {
+ try {
+ fileChannel.close();
+ } catch (IOException e) {
+ ExceptionUtils.rethrow(e, "Failed to close file channel.");
+ }
+ }
+ IOUtils.deleteFileQuietly(dataFilePath);
+ }
+
+ /**
+ * Initialize the file channel in a lazy manner, which can reduce usage of the file descriptor
+ * resource.
+ *
+ * @param partitionId denotes the partition id.
+ * @throws IOException is thrown in the event of an error.
+ */
+ private void lazyInitializeFileChannel(TieredStoragePartitionId partitionId)
+ throws IOException {
+ if (fileChannel == null) {
+ try {
+ fileChannel = FileChannel.open(dataFilePath, StandardOpenOption.READ);
+ } catch (FileNotFoundException e) {
+ throw new PartitionNotFoundException(
+ TieredStorageIdMappingUtils.convertId(partitionId));
+ }
+ }
+ }
+
+ /**
+ * Try to get the region cache associated with the given cache key.
+ *
+ * <p>If the relevant region cache exists within the region caches, it will be returned and
+ * subsequently removed. However, if the region cache does not exist, a new region cache will be
+ * created using the data index and returned.
+ *
+ * @param cacheKey denotes the key of the desired region cache.
+ * @return returns the relevant region cache if it exists, otherwise a null value.
+ * @throws IOException is thrown in the event of an error.
+ */
+ private RegionCache tryGetRegionCache(Tuple2<TieredStorageSubpartitionId, Integer> cacheKey)
+ throws IOException {
+ RegionCache regionCache = regionCaches.remove(cacheKey);
+ if (regionCache == null) {
+ Optional<Region> region = dataIndex.getRegion(cacheKey.f0, cacheKey.f1);
+ if (region.isPresent()) {
+ regionCache = new RegionCache(cacheKey.f1, region.get());
+ }
Review Comment:
Fixed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/ProducerMergedPartitionFileReader.java:
##########
@@ -60,9 +141,122 @@ public long getPriority(
TieredStorageSubpartitionId subpartitionId,
int segmentId,
int bufferIndex) {
- return 0;
+ Tuple2<TieredStorageSubpartitionId, Integer> cacheKey =
+ Tuple2.of(subpartitionId, bufferIndex);
+ RegionCache regionCache = regionCaches.get(cacheKey);
+ return regionCache == null ? Long.MAX_VALUE : regionCache.getFileOffset();
}
@Override
- public void release() {}
+ public void release() {
+ if (fileChannel != null) {
+ try {
+ fileChannel.close();
+ } catch (IOException e) {
+ ExceptionUtils.rethrow(e, "Failed to close file channel.");
+ }
+ }
+ IOUtils.deleteFileQuietly(dataFilePath);
+ }
+
+ /**
+ * Initialize the file channel in a lazy manner, which can reduce usage of the file descriptor
+ * resource.
+ *
+ * @param partitionId denotes the partition id.
+ * @throws IOException is thrown in the event of an error.
+ */
+ private void lazyInitializeFileChannel(TieredStoragePartitionId partitionId)
+ throws IOException {
+ if (fileChannel == null) {
+ try {
+ fileChannel = FileChannel.open(dataFilePath, StandardOpenOption.READ);
+ } catch (FileNotFoundException e) {
+ throw new PartitionNotFoundException(
+ TieredStorageIdMappingUtils.convertId(partitionId));
+ }
+ }
+ }
+
+ /**
+ * Try to get the region cache associated with the given cache key.
+ *
+ * <p>If the relevant region cache exists within the region caches, it will be returned and
+ * subsequently removed. However, if the region cache does not exist, a new region cache will be
+ * created using the data index and returned.
+ *
+ * @param cacheKey denotes the key of the desired region cache.
+ * @return returns the relevant region cache if it exists, otherwise a null value.
+ * @throws IOException is thrown in the event of an error.
+ */
+ private RegionCache tryGetRegionCache(Tuple2<TieredStorageSubpartitionId, Integer> cacheKey)
+ throws IOException {
+ RegionCache regionCache = regionCaches.remove(cacheKey);
+ if (regionCache == null) {
+ Optional<Region> region = dataIndex.getRegion(cacheKey.f0, cacheKey.f1);
+ if (region.isPresent()) {
+ regionCache = new RegionCache(cacheKey.f1, region.get());
+ }
+ } else {
+ numRegionCache--;
+ }
+ return regionCache;
+ }
+
+ /**
+ * The {@link RegionCache} represents the cache of a {@link Region}. Each region cache contains
+ * the last consumed {@link Region}, the next buffer index to consume within the region, and the
+ * file offset of the buffer index.
+ */
+ private class RegionCache {
+
+ private final Region region;
+
+ private int nextBufferIndex;
+
+ private long fileOffset;
+
+ private RegionCache(int bufferIndex, Region region) throws IOException {
+ this.nextBufferIndex = bufferIndex;
+ this.region = region;
+ moveFileOffsetToBuffer(bufferIndex);
+ }
+
+ /**
+ * Get the file offset of next buffer.
+ *
+ * @return the file offset.
+ */
+ private long getFileOffset() {
+ return fileOffset;
+ }
+
+ /**
+ * Updates the {@link RegionCache} upon the retrieval of a buffer from the file using the
+ * file offset in the {@link RegionCache}.
+ *
+ * @param bufferSize denotes the size of the buffer.
+ * @return returns a boolean value indicating the presence or absence of residual buffers in
+ * the region.
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org