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.
- Loading branch information
1 parent
4d158c7
commit 4c1f30b
Showing
5 changed files
with
171 additions
and
26 deletions.
There are no files selected for viewing
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
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, 20, TimeUnit.SECONDS); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
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