You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Zakelly (via GitHub)" <gi...@apache.org> on 2025/02/28 15:59:13 UTC
[PR] [FLINK-37379] Double list LRU in ForSt state and cache reload [flink]
Zakelly opened a new pull request, #26237:
URL: https://github.com/apache/flink/pull/26237
## What is the purpose of the change
Currently, the file-based LRU cache lacks of loading files from the disaggregated DFS. Meaning that after the job recovery, there will be no cache available, hence the performance is low over a period of time. It is critical for production availability of disaggregated state.
This PR introduces a new implementation that allows ForSt to manage file-based cache in a double list (hot/cold) and support cache reload after eviction.
## Brief change log
- Introduce `DoubleListLru` and rewrite the `FileBasedCache` based on that.
## Verifying this change
This change is already covered by existing IT tests. The new introduced classes are covered by new introduced UTs.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #26237:
URL: https://github.com/apache/flink/pull/26237#issuecomment-2694631063
Also, additional two commits to enhance the user experience are attached.
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26237:
URL: https://github.com/apache/flink/pull/26237#discussion_r1977020363
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/DoubleListLru.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.fs.cache;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * A double link LRU (Least Recently Used) cache implementation. This cache maintains two linked
+ * lists to manage the cache entries. The first list contains the most recently used entries, and
+ * the second list contains the less recently used entries. The cache also maintains a middle
+ * pointer to efficiently manage the entries. No thread-safe guarantees are provided.
+ *
+ * @param <K> the type of keys maintained by this cache
+ * @param <V> the type of mapped values
+ */
+public abstract class DoubleListLru<K, V> implements Iterable<Tuple2<K, V>> {
+
+ class Node {
+ K key;
+ V value;
+ Node prev;
+ Node next;
+ boolean isBeforeMiddle;
+
+ Node(K key, V value) {
+ this.key = key;
+ this.value = value;
+ this.isBeforeMiddle = false;
+ }
+ }
+
+ private Node head;
+ private Node tail;
+ private Node middle;
+ private int size;
+ private int secondSize;
+ private final HashMap<K, Node> map;
+
+ public DoubleListLru() {
+ this.head = null;
+ this.tail = null;
+ this.middle = null;
+ this.size = 0;
+ this.map = new HashMap<>();
+ }
+
+ // -------------------
+ // Hook methods
+ // -------------------
+
+ /**
+ * Checks if it is safe to add a value to the first list. Will be called before adding a value
+ * to the first list.
+ *
+ * @param value the value to be added to the first list
+ * @return true if it is safe to add the value to the first list, false otherwise.
+ */
+ abstract boolean isSafeToAddFirst(V value);
+
+ /**
+ * Hook method called when a new node is created.
+ *
+ * @param value the value of the new node
+ * @param n the new node
+ */
+ abstract void newNodeCreated(V value, Node n);
+
+ /**
+ * Hook method called when a value is added to the first list.
+ *
+ * @param value the value added to the first list
+ */
+ abstract void addedToFirst(V value);
+
+ /**
+ * Hook method called when a value is added to the second list.
+ *
+ * @param value the value added to the second list
+ */
+ abstract void addedToSecond(V value);
+
+ /**
+ * Hook method called when a value is removed from the first list.
+ *
+ * @param value the value removed from the first list
+ */
+ abstract void removedFromFirst(V value);
+
+ /**
+ * Hook method called when a value is removed from the second list.
+ *
+ * @param value the value removed from the second list
+ */
+ abstract void removedFromSecond(V value);
+
+ /**
+ * Hook method called when a value is moved to the first list.
+ *
+ * @param value the value moved to the first list
+ */
+ abstract void movedToFirst(V value);
+
+ /**
+ * Hook method called when a value is moved to the second list.
+ *
+ * @param value the value moved to the second list
+ */
+ abstract void movedToSecond(V value);
+
+ /**
+ * Hook method called when a node is accessed in the second list.
+ *
+ * @param value the value of the accessed node
+ * @return true if the node should be promoted to the first list, false otherwise
+ */
+ abstract boolean nodeAccessedAtSecond(V value);
+
+ /**
+ * Hook method called when a value is promoted to the first list.
+ *
+ * @param value the promoted value
+ */
+ abstract void promotedToFirst(V value);
+
+ /**
+ * Adds a new entry to the front of the cache.
+ *
+ * @param key the key of the entry
+ * @param value the value of the entry
+ */
+ public void addFirst(K key, V value) {
+ if (!isSafeToAddFirst(value)) {
+ addSecond(key, value);
+ return;
+ }
+ Node newNode = new Node(key, value);
+ newNodeCreated(value, newNode);
+ map.put(key, newNode);
Review Comment:
No we don't support different value for one key. the cache key is from the source path of MappingEntry, so IIUC in rescaling scenario, we still have one-to-one mapping from key to value here.
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on PR #26237:
URL: https://github.com/apache/flink/pull/26237#issuecomment-2696072001
Thanks for the update, `ForStFlinkFileSystemTest#testSstFileInCache ` failed because we reserved the space.
The design and implementation are fine for me.
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26237:
URL: https://github.com/apache/flink/pull/26237#discussion_r1977016840
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/FileBasedCache.java:
##########
@@ -123,28 +206,185 @@ public void delete(Path path) {
}
}
+ // -----------------------------------------------------------------------
+ // Overriding methods of {@link DoubleListLru} to provide thread-safe.
+ // -----------------------------------------------------------------------
+
@Override
- FileCacheEntry internalGet(String key, FileCacheEntry value) {
- if (metricGroup != null) {
- if (value != null) {
- hitCounter.inc();
- } else {
- missCounter.inc();
+ public FileCacheEntry get(String key, boolean affectOrder) {
+ synchronized (this) {
+ return super.get(key, affectOrder);
+ }
+ }
+
+ @Override
+ public void addFirst(String key, FileCacheEntry value) {
+ synchronized (this) {
+ super.addFirst(key, value);
+ }
+ }
+
+ @Override
+ public void addSecond(String key, FileCacheEntry value) {
+ synchronized (this) {
+ super.addSecond(key, value);
+ }
+ }
+
+ @Override
+ public FileCacheEntry remove(String key) {
+ synchronized (this) {
+ return super.remove(key);
+ }
+ }
+
+ /** Directly insert in cache when restoring. */
+ public void registerInCache(Path originalPath, long size) {
+ Path cachePath = getCachePath(originalPath);
+ FileCacheEntry fileCacheEntry = new FileCacheEntry(this, originalPath, cachePath, size);
+ // We want the new registered cache to load ASAP, so assign a initial access count.
+ fileCacheEntry.accessCountInColdLink = Math.max(0, accessBeforePromote - 2);
+ addSecond(cachePath.toString(), fileCacheEntry);
+ }
+
+ void removeFile(FileCacheEntry entry) {
+ if (closed) {
+ entry.doRemoveFile();
+ } else {
+ executorService.submit(entry::doRemoveFile);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ executorService.shutdown();
+ for (Tuple2<String, FileCacheEntry> entry : this) {
+ entry.f1.close();
+ }
+ }
+
+ // -----------------------------
+ // Hook methods implementation
+ // -----------------------------
+
+ @Override
+ boolean isSafeToAddFirst(FileCacheEntry value) {
+ return cacheLimitPolicy.isSafeToAdd(value.entrySize);
+ }
+
+ @Override
+ void newNodeCreated(FileCacheEntry value, DoubleListLru<String, FileCacheEntry>.Node n) {
+ value.setTouchFunction(
+ () -> {
+ // provide synchronized access to the LRU cache.
+ synchronized (FileBasedCache.this) {
+ accessNode(n);
+ }
+ });
+ }
+
+ @Override
+ void addedToFirst(FileCacheEntry value) {
+ LOG.trace("Cache entry {} added to first link.", value.cachePath);
+ while (cacheLimitPolicy.isOverflow(value.entrySize)) {
+ moveMiddleFront();
+ }
+ cacheLimitPolicy.acquire(value.entrySize);
+ }
+
+ @Override
+ void addedToSecond(FileCacheEntry value) {
+ LOG.trace("Cache entry {} added to second link.", value.cachePath);
+ value.secondAccessEpoch = (++secondAccessEpoch);
+ }
+
+ @Override
+ void removedFromFirst(FileCacheEntry value) {
+ cacheLimitPolicy.release(value.entrySize);
+ value.close();
+ }
+
+ @Override
+ void removedFromSecond(FileCacheEntry value) {
+ value.close();
+ }
+
+ @Override
+ void movedToFirst(FileCacheEntry entry) {
+ // here we won't consider the cache limit policy.
+ // since there will be promotedToFirst called after this.
+ LOG.trace("Cache entry {} moved to first link.", entry.cachePath);
+ // trigger the loading
+ if (entry.switchStatus(
+ FileCacheEntry.EntryStatus.INVALID, FileCacheEntry.EntryStatus.LOADED)) {
+ // just a try
+ entry.loaded();
+ if (loadBackCounter != null) {
+ loadBackCounter.inc();
}
}
- return value;
+ if (entry.switchStatus(
+ FileCacheEntry.EntryStatus.REMOVED, FileCacheEntry.EntryStatus.LOADING)) {
+ executorService.submit(
+ () -> {
+ if (entry.checkStatus(FileCacheEntry.EntryStatus.LOADING)) {
+ Path path = entry.load();
+ if (path == null) {
+ entry.switchStatus(
+ FileCacheEntry.EntryStatus.LOADING,
+ FileCacheEntry.EntryStatus.REMOVED);
+ } else if (entry.switchStatus(
+ FileCacheEntry.EntryStatus.LOADING,
+ FileCacheEntry.EntryStatus.LOADED)) {
+ entry.loaded();
+ if (loadBackCounter != null) {
+ loadBackCounter.inc();
+ }
+ } else {
+ try {
+ path.getFileSystem().delete(path, false);
+ // delete the file
+ } catch (IOException e) {
+ }
+ }
+ }
+ });
+ }
}
@Override
- void internalInsert(String key, FileCacheEntry value) {}
+ void movedToSecond(FileCacheEntry value) {
+ // trigger the evicting
+ LOG.trace("Cache entry {} moved to second link.", value.cachePath);
+ cacheLimitPolicy.release(value.entrySize);
Review Comment:
The cache will be deleted in `value.invalidate()`, if the reference count reached 0.
We'd better not wait for that, since we are on the task thread and it's better not block here.
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26237:
URL: https://github.com/apache/flink/pull/26237#discussion_r1978510149
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataOutputStream.java:
##########
@@ -64,58 +66,80 @@ public CachedDataOutputStream(
@Override
public long getPos() throws IOException {
- return cacheOutputStream.getPos();
+ if (cacheOutputStream == null) {
+ return originOutputStream.getPos();
+ } else {
+ return cacheOutputStream.getPos();
+ }
}
@Override
public void write(int b) throws IOException {
- cacheOutputStream.write(b);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b);
+ }
originOutputStream.write(b);
}
public void write(byte[] b) throws IOException {
- cacheOutputStream.write(b);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b);
+ }
originOutputStream.write(b);
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
- cacheOutputStream.write(b, off, len);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b, off, len);
+ }
originOutputStream.write(b, off, len);
}
@Override
public void flush() throws IOException {
- cacheOutputStream.flush();
+ if (cacheOutputStream != null) {
+ cacheOutputStream.flush();
+ }
originOutputStream.flush();
}
@Override
public void sync() throws IOException {
- cacheOutputStream.sync();
+ if (cacheOutputStream != null) {
+ cacheOutputStream.sync();
+ }
originOutputStream.sync();
}
@Override
public void close() throws IOException {
+ long size = getPos();
if (originOutputStream != null) {
originOutputStream.close();
originOutputStream = null;
}
if (cacheOutputStream != null) {
- putIntoCache();
+ putIntoCache(size);
cacheOutputStream.close();
cacheOutputStream = null;
+ } else {
+ registerIntoCache(size);
}
}
- private void putIntoCache() throws IOException {
- long thisSize = cacheOutputStream.getPos();
+ private void putIntoCache(long size) {
FileCacheEntry fileCacheEntry =
- new FileCacheEntry(fileBasedCache, originalPath, cachePath, thisSize);
+ new FileCacheEntry(fileBasedCache, originalPath, cachePath, size);
fileCacheEntry.switchStatus(
FileCacheEntry.EntryStatus.REMOVED, FileCacheEntry.EntryStatus.LOADED);
fileCacheEntry.loaded();
fileBasedCache.addFirst(cachePath.toString(), fileCacheEntry);
}
+
+ private void registerIntoCache(long size) {
+ FileCacheEntry fileCacheEntry =
Review Comment:
Switch Status REMOVED->INVALID?
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26237:
URL: https://github.com/apache/flink/pull/26237#discussion_r1978538199
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataOutputStream.java:
##########
@@ -64,58 +66,80 @@ public CachedDataOutputStream(
@Override
public long getPos() throws IOException {
- return cacheOutputStream.getPos();
+ if (cacheOutputStream == null) {
+ return originOutputStream.getPos();
+ } else {
+ return cacheOutputStream.getPos();
+ }
}
@Override
public void write(int b) throws IOException {
- cacheOutputStream.write(b);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b);
+ }
originOutputStream.write(b);
}
public void write(byte[] b) throws IOException {
- cacheOutputStream.write(b);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b);
+ }
originOutputStream.write(b);
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
- cacheOutputStream.write(b, off, len);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b, off, len);
+ }
originOutputStream.write(b, off, len);
}
@Override
public void flush() throws IOException {
- cacheOutputStream.flush();
+ if (cacheOutputStream != null) {
+ cacheOutputStream.flush();
+ }
originOutputStream.flush();
}
@Override
public void sync() throws IOException {
- cacheOutputStream.sync();
+ if (cacheOutputStream != null) {
+ cacheOutputStream.sync();
+ }
originOutputStream.sync();
}
@Override
public void close() throws IOException {
+ long size = getPos();
if (originOutputStream != null) {
originOutputStream.close();
originOutputStream = null;
}
if (cacheOutputStream != null) {
- putIntoCache();
+ putIntoCache(size);
cacheOutputStream.close();
cacheOutputStream = null;
+ } else {
+ registerIntoCache(size);
}
}
- private void putIntoCache() throws IOException {
- long thisSize = cacheOutputStream.getPos();
+ private void putIntoCache(long size) {
FileCacheEntry fileCacheEntry =
- new FileCacheEntry(fileBasedCache, originalPath, cachePath, thisSize);
+ new FileCacheEntry(fileBasedCache, originalPath, cachePath, size);
fileCacheEntry.switchStatus(
FileCacheEntry.EntryStatus.REMOVED, FileCacheEntry.EntryStatus.LOADED);
fileCacheEntry.loaded();
fileBasedCache.addFirst(cachePath.toString(), fileCacheEntry);
}
+
+ private void registerIntoCache(long size) {
+ FileCacheEntry fileCacheEntry =
Review Comment:
👌thanks for the explanation.
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26237:
URL: https://github.com/apache/flink/pull/26237#issuecomment-2691091645
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c2553de448445dca90187b1283cc1c831c73fec8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c2553de448445dca90187b1283cc1c831c73fec8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c2553de448445dca90187b1283cc1c831c73fec8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #26237:
URL: https://github.com/apache/flink/pull/26237#discussion_r1976945810
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/FileBasedCache.java:
##########
@@ -123,28 +206,185 @@ public void delete(Path path) {
}
}
+ // -----------------------------------------------------------------------
+ // Overriding methods of {@link DoubleListLru} to provide thread-safe.
+ // -----------------------------------------------------------------------
+
@Override
- FileCacheEntry internalGet(String key, FileCacheEntry value) {
- if (metricGroup != null) {
- if (value != null) {
- hitCounter.inc();
- } else {
- missCounter.inc();
+ public FileCacheEntry get(String key, boolean affectOrder) {
+ synchronized (this) {
+ return super.get(key, affectOrder);
+ }
+ }
+
+ @Override
+ public void addFirst(String key, FileCacheEntry value) {
+ synchronized (this) {
+ super.addFirst(key, value);
+ }
+ }
+
+ @Override
+ public void addSecond(String key, FileCacheEntry value) {
+ synchronized (this) {
+ super.addSecond(key, value);
+ }
+ }
+
+ @Override
+ public FileCacheEntry remove(String key) {
+ synchronized (this) {
+ return super.remove(key);
+ }
+ }
+
+ /** Directly insert in cache when restoring. */
+ public void registerInCache(Path originalPath, long size) {
+ Path cachePath = getCachePath(originalPath);
+ FileCacheEntry fileCacheEntry = new FileCacheEntry(this, originalPath, cachePath, size);
+ // We want the new registered cache to load ASAP, so assign a initial access count.
+ fileCacheEntry.accessCountInColdLink = Math.max(0, accessBeforePromote - 2);
+ addSecond(cachePath.toString(), fileCacheEntry);
+ }
+
+ void removeFile(FileCacheEntry entry) {
+ if (closed) {
+ entry.doRemoveFile();
+ } else {
+ executorService.submit(entry::doRemoveFile);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ executorService.shutdown();
+ for (Tuple2<String, FileCacheEntry> entry : this) {
+ entry.f1.close();
+ }
+ }
+
+ // -----------------------------
+ // Hook methods implementation
+ // -----------------------------
+
+ @Override
+ boolean isSafeToAddFirst(FileCacheEntry value) {
+ return cacheLimitPolicy.isSafeToAdd(value.entrySize);
+ }
+
+ @Override
+ void newNodeCreated(FileCacheEntry value, DoubleListLru<String, FileCacheEntry>.Node n) {
+ value.setTouchFunction(
+ () -> {
+ // provide synchronized access to the LRU cache.
+ synchronized (FileBasedCache.this) {
+ accessNode(n);
+ }
+ });
+ }
+
+ @Override
+ void addedToFirst(FileCacheEntry value) {
+ LOG.trace("Cache entry {} added to first link.", value.cachePath);
+ while (cacheLimitPolicy.isOverflow(value.entrySize)) {
+ moveMiddleFront();
+ }
+ cacheLimitPolicy.acquire(value.entrySize);
+ }
+
+ @Override
+ void addedToSecond(FileCacheEntry value) {
+ LOG.trace("Cache entry {} added to second link.", value.cachePath);
+ value.secondAccessEpoch = (++secondAccessEpoch);
+ }
+
+ @Override
+ void removedFromFirst(FileCacheEntry value) {
+ cacheLimitPolicy.release(value.entrySize);
+ value.close();
+ }
+
+ @Override
+ void removedFromSecond(FileCacheEntry value) {
+ value.close();
+ }
+
+ @Override
+ void movedToFirst(FileCacheEntry entry) {
+ // here we won't consider the cache limit policy.
+ // since there will be promotedToFirst called after this.
+ LOG.trace("Cache entry {} moved to first link.", entry.cachePath);
+ // trigger the loading
+ if (entry.switchStatus(
+ FileCacheEntry.EntryStatus.INVALID, FileCacheEntry.EntryStatus.LOADED)) {
+ // just a try
+ entry.loaded();
+ if (loadBackCounter != null) {
+ loadBackCounter.inc();
}
}
- return value;
+ if (entry.switchStatus(
+ FileCacheEntry.EntryStatus.REMOVED, FileCacheEntry.EntryStatus.LOADING)) {
+ executorService.submit(
+ () -> {
+ if (entry.checkStatus(FileCacheEntry.EntryStatus.LOADING)) {
+ Path path = entry.load();
+ if (path == null) {
+ entry.switchStatus(
+ FileCacheEntry.EntryStatus.LOADING,
+ FileCacheEntry.EntryStatus.REMOVED);
+ } else if (entry.switchStatus(
+ FileCacheEntry.EntryStatus.LOADING,
+ FileCacheEntry.EntryStatus.LOADED)) {
+ entry.loaded();
+ if (loadBackCounter != null) {
+ loadBackCounter.inc();
+ }
+ } else {
+ try {
+ path.getFileSystem().delete(path, false);
+ // delete the file
+ } catch (IOException e) {
+ }
+ }
+ }
+ });
+ }
}
@Override
- void internalInsert(String key, FileCacheEntry value) {}
+ void movedToSecond(FileCacheEntry value) {
+ // trigger the evicting
+ LOG.trace("Cache entry {} moved to second link.", value.cachePath);
+ cacheLimitPolicy.release(value.entrySize);
Review Comment:
Is the `FileCacheEntry#cachePath` deleted from `cacheFs`?
Do we need to wait for delete before releasing `cacheLimitPolicy`?
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/DoubleListLru.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst.fs.cache;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * A double link LRU (Least Recently Used) cache implementation. This cache maintains two linked
+ * lists to manage the cache entries. The first list contains the most recently used entries, and
+ * the second list contains the less recently used entries. The cache also maintains a middle
+ * pointer to efficiently manage the entries. No thread-safe guarantees are provided.
+ *
+ * @param <K> the type of keys maintained by this cache
+ * @param <V> the type of mapped values
+ */
+public abstract class DoubleListLru<K, V> implements Iterable<Tuple2<K, V>> {
+
+ class Node {
+ K key;
+ V value;
+ Node prev;
+ Node next;
+ boolean isBeforeMiddle;
+
+ Node(K key, V value) {
+ this.key = key;
+ this.value = value;
+ this.isBeforeMiddle = false;
+ }
+ }
+
+ private Node head;
+ private Node tail;
+ private Node middle;
+ private int size;
+ private int secondSize;
+ private final HashMap<K, Node> map;
+
+ public DoubleListLru() {
+ this.head = null;
+ this.tail = null;
+ this.middle = null;
+ this.size = 0;
+ this.map = new HashMap<>();
+ }
+
+ // -------------------
+ // Hook methods
+ // -------------------
+
+ /**
+ * Checks if it is safe to add a value to the first list. Will be called before adding a value
+ * to the first list.
+ *
+ * @param value the value to be added to the first list
+ * @return true if it is safe to add the value to the first list, false otherwise.
+ */
+ abstract boolean isSafeToAddFirst(V value);
+
+ /**
+ * Hook method called when a new node is created.
+ *
+ * @param value the value of the new node
+ * @param n the new node
+ */
+ abstract void newNodeCreated(V value, Node n);
+
+ /**
+ * Hook method called when a value is added to the first list.
+ *
+ * @param value the value added to the first list
+ */
+ abstract void addedToFirst(V value);
+
+ /**
+ * Hook method called when a value is added to the second list.
+ *
+ * @param value the value added to the second list
+ */
+ abstract void addedToSecond(V value);
+
+ /**
+ * Hook method called when a value is removed from the first list.
+ *
+ * @param value the value removed from the first list
+ */
+ abstract void removedFromFirst(V value);
+
+ /**
+ * Hook method called when a value is removed from the second list.
+ *
+ * @param value the value removed from the second list
+ */
+ abstract void removedFromSecond(V value);
+
+ /**
+ * Hook method called when a value is moved to the first list.
+ *
+ * @param value the value moved to the first list
+ */
+ abstract void movedToFirst(V value);
+
+ /**
+ * Hook method called when a value is moved to the second list.
+ *
+ * @param value the value moved to the second list
+ */
+ abstract void movedToSecond(V value);
+
+ /**
+ * Hook method called when a node is accessed in the second list.
+ *
+ * @param value the value of the accessed node
+ * @return true if the node should be promoted to the first list, false otherwise
+ */
+ abstract boolean nodeAccessedAtSecond(V value);
+
+ /**
+ * Hook method called when a value is promoted to the first list.
+ *
+ * @param value the promoted value
+ */
+ abstract void promotedToFirst(V value);
+
+ /**
+ * Adds a new entry to the front of the cache.
+ *
+ * @param key the key of the entry
+ * @param value the value of the entry
+ */
+ public void addFirst(K key, V value) {
+ if (!isSafeToAddFirst(value)) {
+ addSecond(key, value);
+ return;
+ }
+ Node newNode = new Node(key, value);
+ newNodeCreated(value, newNode);
+ map.put(key, newNode);
Review Comment:
Can the same key have multiple different values when rescaling down?
Will the following value overwrite the previous value?
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #26237:
URL: https://github.com/apache/flink/pull/26237#discussion_r1978519401
##########
flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataOutputStream.java:
##########
@@ -64,58 +66,80 @@ public CachedDataOutputStream(
@Override
public long getPos() throws IOException {
- return cacheOutputStream.getPos();
+ if (cacheOutputStream == null) {
+ return originOutputStream.getPos();
+ } else {
+ return cacheOutputStream.getPos();
+ }
}
@Override
public void write(int b) throws IOException {
- cacheOutputStream.write(b);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b);
+ }
originOutputStream.write(b);
}
public void write(byte[] b) throws IOException {
- cacheOutputStream.write(b);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b);
+ }
originOutputStream.write(b);
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
- cacheOutputStream.write(b, off, len);
+ if (cacheOutputStream != null) {
+ cacheOutputStream.write(b, off, len);
+ }
originOutputStream.write(b, off, len);
}
@Override
public void flush() throws IOException {
- cacheOutputStream.flush();
+ if (cacheOutputStream != null) {
+ cacheOutputStream.flush();
+ }
originOutputStream.flush();
}
@Override
public void sync() throws IOException {
- cacheOutputStream.sync();
+ if (cacheOutputStream != null) {
+ cacheOutputStream.sync();
+ }
originOutputStream.sync();
}
@Override
public void close() throws IOException {
+ long size = getPos();
if (originOutputStream != null) {
originOutputStream.close();
originOutputStream = null;
}
if (cacheOutputStream != null) {
- putIntoCache();
+ putIntoCache(size);
cacheOutputStream.close();
cacheOutputStream = null;
+ } else {
+ registerIntoCache(size);
}
}
- private void putIntoCache() throws IOException {
- long thisSize = cacheOutputStream.getPos();
+ private void putIntoCache(long size) {
FileCacheEntry fileCacheEntry =
- new FileCacheEntry(fileBasedCache, originalPath, cachePath, thisSize);
+ new FileCacheEntry(fileBasedCache, originalPath, cachePath, size);
fileCacheEntry.switchStatus(
FileCacheEntry.EntryStatus.REMOVED, FileCacheEntry.EntryStatus.LOADED);
fileCacheEntry.loaded();
fileBasedCache.addFirst(cachePath.toString(), fileCacheEntry);
}
+
+ private void registerIntoCache(long size) {
+ FileCacheEntry fileCacheEntry =
Review Comment:
It should be `REMOVED`, since it has no file.
--
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
Re: [PR] [FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly merged PR #26237:
URL: https://github.com/apache/flink/pull/26237
--
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