forked from apache/doris
-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Fix](hdfs-fs)The cache expiration should explicitly release the held…
… fs (apache#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. (cherry picked from commit 922ec3a)
- Loading branch information
1 parent
d44ee1c
commit c390ee9
Showing
5 changed files
with
197 additions
and
1 deletion.
There are no files selected for viewing
117 changes: 117 additions & 0 deletions
117
fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFSPhantomManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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); | ||
} | ||
} | ||
} | ||
} | ||
|
||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
44 changes: 44 additions & 0 deletions
44
fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystemPhantomReference.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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; | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters