This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit bc02f3f8008374b79c2622d9196bf89c09cd9a90
Author: Calvin Kirs <k...@apache.org>
AuthorDate: Fri Sep 6 12:31:00 2024 +0800
[Fix](hdfs-fs)The cache expiration should explicitly release the held fs
(#38610)
## Proposed changes
The RemoteFSPhantomManager class is responsible for managing phantom
references of RemoteFileSystem objects, ensuring that the associated
FileSystem resources are automatically cleaned up when RemoteFileSystem
objects are garbage collected.
Key features:
- Phantom Reference Monitoring: The class uses a ReferenceQueue and
PhantomReference to track RemoteFileSystem objects. When these objects
are no longer in use and garbage collected, the class ensures the
corresponding FileSystem resources are properly closed to prevent
resource leaks.
- Thread-safe Cleanup: It provides a thread-safe mechanism to start a
cleanup thread only once. This thread runs periodically, checking the
ReferenceQueue and closing any unused FileSystem resources.
Resource Management: The class maintains a map between phantom
references and their corresponding FileSystem objects, ensuring that
these resources are cleaned up appropriately.
- The cleanup thread runs at regular intervals, ensuring that any
RemoteFileSystem object that is no longer in use is safely removed along
with its associated FileSystem resources.
---
.../doris/fs/remote/RemoteFSPhantomManager.java | 117 +++++++++++++++++++++
.../apache/doris/fs/remote/RemoteFileSystem.java | 21 +++-
.../remote/RemoteFileSystemPhantomReference.java | 44 ++++++++
.../org/apache/doris/fs/remote/S3FileSystem.java | 8 ++
.../apache/doris/fs/remote/dfs/DFSFileSystem.java | 8 ++
5 files changed, 197 insertions(+), 1 deletion(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFSPhantomManager.java
b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFSPhantomManager.java
new file mode 100644
index 00000000000..282361c4cb6
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFSPhantomManager.java
@@ -0,0 +1,117 @@
+// 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.doris.fs.remote;
+
+import org.apache.doris.common.CustomThreadFactory;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.IOException;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.Reference;
+import java.lang.ref.ReferenceQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * The RemoteFSPhantomManager class is responsible for managing the phantom
references
+ * of RemoteFileSystem objects. It ensures that the associated FileSystem
resources are
+ * automatically cleaned up when the RemoteFileSystem objects are garbage
collected.
+ * <p>
+ * By utilizing a ReferenceQueue and PhantomReference, this class can monitor
the lifecycle
+ * of RemoteFileSystem objects. When a RemoteFileSystem object is no longer in
use and is
+ * garbage collected, its corresponding FileSystem resource is properly closed
to prevent
+ * resource leaks.
+ * <p>
+ * The class provides a thread-safe mechanism to ensure that the cleanup
thread is started only once.
+ * <p>
+ * Main functionalities include:
+ * - Registering phantom references of RemoteFileSystem objects.
+ * - Starting a periodic cleanup thread that automatically closes unused
FileSystem resources.
+ */
+public class RemoteFSPhantomManager {
+
+ private static final Logger LOG =
LogManager.getLogger(RemoteFSPhantomManager.class);
+
+ // Scheduled executor for periodic resource cleanup
+ private static ScheduledExecutorService cleanupExecutor;
+
+ // Reference queue for monitoring RemoteFileSystem objects' phantom
references
+ private static final ReferenceQueue<RemoteFileSystem> referenceQueue = new
ReferenceQueue<>();
+
+ // Map storing the phantom references and their corresponding FileSystem
objects
+ private static final ConcurrentHashMap<PhantomReference<RemoteFileSystem>,
FileSystem> referenceMap
+ = new ConcurrentHashMap<>();
+
+ // Flag indicating whether the cleanup thread has been started
+ private static final AtomicBoolean isStarted = new AtomicBoolean(false);
+
+ /**
+ * Registers a phantom reference for a RemoteFileSystem object in the
manager.
+ * If the cleanup thread has not been started, it will be started.
+ *
+ * @param remoteFileSystem the RemoteFileSystem object to be registered
+ */
+ public static void registerPhantomReference(RemoteFileSystem
remoteFileSystem) {
+ if (!isStarted.get()) {
+ start();
+ isStarted.set(true);
+ }
+ RemoteFileSystemPhantomReference phantomReference = new
RemoteFileSystemPhantomReference(remoteFileSystem,
+ referenceQueue);
+ referenceMap.put(phantomReference, remoteFileSystem.dfsFileSystem);
+ }
+
+ /**
+ * Starts the cleanup thread, which periodically checks and cleans up
unused FileSystem resources.
+ * The method uses double-checked locking to ensure thread-safe startup of
the cleanup thread.
+ */
+ public static void start() {
+ if (isStarted.compareAndSet(false, true)) {
+ synchronized (RemoteFSPhantomManager.class) {
+ LOG.info("Starting cleanup thread for RemoteFileSystem
objects");
+ if (cleanupExecutor == null) {
+ CustomThreadFactory threadFactory = new
CustomThreadFactory("remote-fs-phantom-cleanup");
+ cleanupExecutor = Executors.newScheduledThreadPool(1,
threadFactory);
+ cleanupExecutor.scheduleAtFixedRate(() -> {
+ Reference<? extends RemoteFileSystem> ref;
+ while ((ref = referenceQueue.poll()) != null) {
+ RemoteFileSystemPhantomReference phantomRef =
(RemoteFileSystemPhantomReference) ref;
+
+ FileSystem fs = referenceMap.remove(phantomRef);
+ if (fs != null) {
+ try {
+ fs.close();
+ LOG.info("Closed file system: {}",
fs.getUri());
+ } catch (IOException e) {
+ LOG.warn("Failed to close file system", e);
+ }
+ }
+ }
+ }, 0, 1, TimeUnit.MINUTES);
+ }
+ }
+ }
+ }
+
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystem.java
b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystem.java
index 3cb8a036c2d..08b7e1cde78 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystem.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystem.java
@@ -30,15 +30,20 @@ import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
+import java.io.Closeable;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
-public abstract class RemoteFileSystem extends PersistentFileSystem {
+public abstract class RemoteFileSystem extends PersistentFileSystem implements
Closeable {
// this field will be visited by multi-threads, better use volatile
qualifier
protected volatile org.apache.hadoop.fs.FileSystem dfsFileSystem = null;
+ private final ReentrantLock fsLock = new ReentrantLock();
+ protected static final AtomicBoolean closed = new AtomicBoolean(false);
public RemoteFileSystem(String name, StorageBackend.StorageType type) {
super(name, type);
@@ -120,4 +125,18 @@ public abstract class RemoteFileSystem extends
PersistentFileSystem {
return rename(origFilePath, destFilePath);
}
+
+ @Override
+ public void close() throws IOException {
+ fsLock.lock();
+ try {
+ if (!closed.getAndSet(true)) {
+ if (dfsFileSystem != null) {
+ dfsFileSystem.close();
+ }
+ }
+ } finally {
+ fsLock.unlock();
+ }
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystemPhantomReference.java
b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystemPhantomReference.java
new file mode 100644
index 00000000000..89506c7b212
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystemPhantomReference.java
@@ -0,0 +1,44 @@
+// 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.doris.fs.remote;
+
+import org.apache.hadoop.fs.FileSystem;
+
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
+
+public class RemoteFileSystemPhantomReference extends
PhantomReference<RemoteFileSystem> {
+
+ private FileSystem fs;
+
+ /**
+ * Creates a new phantom reference that refers to the given object and
+ * is registered with the given queue.
+ *
+ * <p> It is possible to create a phantom reference with a {@code null}
+ * queue. Such a reference will never be enqueued.
+ *
+ * @param referent the object the new phantom reference will refer to
+ * @param q the queue with which the reference is to be registered,
+ * or {@code null} if registration is not required
+ */
+ public RemoteFileSystemPhantomReference(RemoteFileSystem referent,
ReferenceQueue<? super RemoteFileSystem> q) {
+ super(referent, q);
+ this.fs = referent.dfsFileSystem;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/fs/remote/S3FileSystem.java
b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/S3FileSystem.java
index 525d80d6797..87ba086baec 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/remote/S3FileSystem.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/S3FileSystem.java
@@ -58,8 +58,15 @@ public class S3FileSystem extends ObjFileSystem {
@Override
protected FileSystem nativeFileSystem(String remotePath) throws
UserException {
+ //todo Extracting a common method to achieve logic reuse
+ if (closed.get()) {
+ throw new UserException("FileSystem is closed.");
+ }
if (dfsFileSystem == null) {
synchronized (this) {
+ if (closed.get()) {
+ throw new UserException("FileSystem is closed.");
+ }
if (dfsFileSystem == null) {
Configuration conf =
DFSFileSystem.getHdfsConf(ifNotSetFallbackToSimpleAuth());
System.setProperty("com.amazonaws.services.s3.enableV4",
"true");
@@ -72,6 +79,7 @@ public class S3FileSystem extends ObjFileSystem {
} catch (Exception e) {
throw new UserException("Failed to get S3 FileSystem
for " + e.getMessage(), e);
}
+ RemoteFSPhantomManager.registerPhantomReference(this);
}
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/fs/remote/dfs/DFSFileSystem.java
b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/dfs/DFSFileSystem.java
index 59fbd73bda7..7034641a9fc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/remote/dfs/DFSFileSystem.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/fs/remote/dfs/DFSFileSystem.java
@@ -26,6 +26,7 @@ import org.apache.doris.common.util.URI;
import org.apache.doris.fs.operations.HDFSFileOperations;
import org.apache.doris.fs.operations.HDFSOpParams;
import org.apache.doris.fs.operations.OpParams;
+import org.apache.doris.fs.remote.RemoteFSPhantomManager;
import org.apache.doris.fs.remote.RemoteFile;
import org.apache.doris.fs.remote.RemoteFileSystem;
@@ -75,8 +76,14 @@ public class DFSFileSystem extends RemoteFileSystem {
@VisibleForTesting
@Override
public FileSystem nativeFileSystem(String remotePath) throws UserException
{
+ if (closed.get()) {
+ throw new UserException("FileSystem is closed.");
+ }
if (dfsFileSystem == null) {
synchronized (this) {
+ if (closed.get()) {
+ throw new UserException("FileSystem is closed.");
+ }
if (dfsFileSystem == null) {
Configuration conf =
getHdfsConf(ifNotSetFallbackToSimpleAuth());
for (Map.Entry<String, String> propEntry :
properties.entrySet()) {
@@ -96,6 +103,7 @@ public class DFSFileSystem extends RemoteFileSystem {
throw new UserException(e);
}
operations = new HDFSFileOperations(dfsFileSystem);
+ RemoteFSPhantomManager.registerPhantomReference(this);
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
