Zakelly commented on code in PR #25561: URL: https://github.com/apache/flink/pull/25561#discussion_r1820533404
########## flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataInputStream.java: ########## @@ -0,0 +1,237 @@ +/* + * 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.core.fs.ByteBufferReadable; +import org.apache.flink.core.fs.FSDataInputStream; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.concurrent.Semaphore; + +/** + * A {@link FSDataInputStream} delegates requests to other one and supports reading data with {@link + * ByteBuffer}. + */ +public class CachedDataInputStream extends FSDataInputStream implements ByteBufferReadable { + + private static final Logger LOG = LoggerFactory.getLogger(CachedDataInputStream.class); + + /** The reference to the cache entry. */ + private final FileCacheEntry cacheEntry; + + private volatile FSDataInputStream fsdis; + + private volatile StreamStatus streamStatus; + + /** + * The position of the cached stream, when cached stream is closed, the position is stored. When + * switch to original stream, the position is restored. + */ + private volatile long position; + + private final FSDataInputStream originalStream; + + private Semaphore semaphore; + + public CachedDataInputStream( + FileCacheEntry cacheEntry, + FSDataInputStream cacheStream, + FSDataInputStream originalStream) { + this.cacheEntry = cacheEntry; + this.fsdis = cacheStream; + this.originalStream = originalStream; + this.streamStatus = StreamStatus.CACHED_OPEN; + this.semaphore = new Semaphore(0); + } + + private FSDataInputStream getStream() throws IOException { + if (streamStatus == StreamStatus.CACHED_OPEN && cacheEntry.tryRetain() > 0) { + return fsdis; + } else if (streamStatus == StreamStatus.CACHED_CLOSED) { Review Comment: ```suggestion } else if (streamStatus != StreamStatus.ORIGINAL) { ``` Since the `streamStatus` might not yet updated ########## flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataInputStream.java: ########## @@ -0,0 +1,237 @@ +/* + * 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.core.fs.ByteBufferReadable; +import org.apache.flink.core.fs.FSDataInputStream; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.concurrent.Semaphore; + +/** + * A {@link FSDataInputStream} delegates requests to other one and supports reading data with {@link + * ByteBuffer}. + */ +public class CachedDataInputStream extends FSDataInputStream implements ByteBufferReadable { Review Comment: Please offer some description about the muti-threading scenario? ########## flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/cache/CachedDataInputStream.java: ########## @@ -0,0 +1,237 @@ +/* + * 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.core.fs.ByteBufferReadable; +import org.apache.flink.core.fs.FSDataInputStream; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.concurrent.Semaphore; + +/** + * A {@link FSDataInputStream} delegates requests to other one and supports reading data with {@link + * ByteBuffer}. + */ +public class CachedDataInputStream extends FSDataInputStream implements ByteBufferReadable { + + private static final Logger LOG = LoggerFactory.getLogger(CachedDataInputStream.class); + + /** The reference to the cache entry. */ + private final FileCacheEntry cacheEntry; + + private volatile FSDataInputStream fsdis; + + private volatile StreamStatus streamStatus; + + /** + * The position of the cached stream, when cached stream is closed, the position is stored. When + * switch to original stream, the position is restored. + */ + private volatile long position; + + private final FSDataInputStream originalStream; + + private Semaphore semaphore; + + public CachedDataInputStream( + FileCacheEntry cacheEntry, + FSDataInputStream cacheStream, + FSDataInputStream originalStream) { + this.cacheEntry = cacheEntry; + this.fsdis = cacheStream; + this.originalStream = originalStream; + this.streamStatus = StreamStatus.CACHED_OPEN; + this.semaphore = new Semaphore(0); + } + + private FSDataInputStream getStream() throws IOException { + if (streamStatus == StreamStatus.CACHED_OPEN && cacheEntry.tryRetain() > 0) { + return fsdis; + } else if (streamStatus == StreamStatus.CACHED_CLOSED) { + try { + semaphore.acquire(1); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + originalStream.seek(position); + position = -1; + streamStatus = StreamStatus.ORIGINAL; + return originalStream; + } else { + return originalStream; + } + } + + private void closeStream() throws IOException { + if (streamStatus == StreamStatus.CACHED_OPEN) { + streamStatus = StreamStatus.CACHED_CLOSED; + position = fsdis.getPos(); + fsdis = null; + semaphore.release(1); + } + } + + private void finish() { + if (streamStatus == StreamStatus.CACHED_OPEN) { Review Comment: The `streamStatus` might be updated after the stream is read. -- 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
