You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by StefanRRichter <gi...@git.apache.org> on 2016/10/25 14:21:06 UTC
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
GitHub user StefanRRichter opened a pull request:
https://github.com/apache/flink/pull/2691
[FLINK-4910] Introduce safety net for closing file system streams
This PR introduces a safety net against leaking/unclosed streams obtained through ``FileSystem``.
The main addition is ``SafetyNetCloseableRegistry``, which tracks all opened streams and uses PhantomReferences to wrapper classes around the streams in order to detect leaking resources after GC of the wrapper class.
Wrapper classes are ``ClosingFSDataInputStream``, ``ClosingFSDataOutputStream``, and ``SafetyNetWrapperFileSystem``, where the latter wraps all created streams as closing and registers them with the ``SafetyNetCloseableRegistry``.
``InheritableThreadLocal<SafetyNetCloseableRegistry>`` is used to scope the registry per Task, where the creation happens at the beginning of a Task's run method and cleanup happens at it's end (and on cancelation).
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/StefanRRichter/flink closeable-io
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2691.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2691
----
commit 384e6d3e3c9640afcbe79d732e34819fcdd44d8e
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-10-24T15:49:54Z
phantom closeable registry
commit 13e719bb7f19d3a4109aa39bdd75a310bdf80bfe
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-10-24T22:14:59Z
Abstractions to avoid code duplication
commit 2e5409c64eff58d28198acf2cab4ea8aa47e8fd1
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-10-24T22:21:48Z
Renames
commit c66049397ebfb1b1297277c518d821e08a1cf36a
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-10-24T23:27:51Z
Logging and removed sync
commit b024717646e2af3abe4f01f00921abd815d5781f
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-10-25T08:44:27Z
Debug messages
commit 020e21e91a193a95f642361463a746a1fd2be67b
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-10-25T11:33:38Z
Integration via Fs Wrapper
commit e86b41575d2b11ca3d492007c746b03ac5907941
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-10-25T12:43:46Z
Proper positioning of calls to register/unregister
commit ca79e84680dcbedeaa1b2583f6bc6a77a97abb43
Author: Stefan Richter <s....@data-artisans.com>
Date: 2016-10-25T13:07:40Z
Documentation and minor fixes
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85329878
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java ---
@@ -446,6 +447,8 @@ public void startTaskThread() {
@Override
public void run() {
+ FileSystem.createFileSystemCloseableRegistryForTask();
--- End diff --
Minor: Can you move that into the beginning of the `try` block? It may be that the task exits early (immediate cancellation) and then we would not have to create this at all.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85570641
--- Diff: flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetCloseableRegistry.java ---
@@ -0,0 +1,181 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.util.AbstractCloseableRegistry;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingProxyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
+import java.util.IdentityHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This implementation of an {@link AbstractCloseableRegistry} registers {@link WrappingProxyCloseable}. When
+ * the proxy becomes subject to GC, this registry takes care of closing unclosed {@link Closeable}s.
+ * <p>
+ * Phantom references are used to track when {@link org.apache.flink.util.WrappingProxy}s of {@link Closeable} got
+ * GC'ed. We ensure that the wrapped {@link Closeable} is proeprly closed to avoid resource leaks.
--- End diff --
typo: properly
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85332853
--- Diff: flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetCloseableRegistry.java ---
@@ -0,0 +1,165 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingProxyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
+import java.util.IdentityHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This implementation of an {@link AbstractCloseableRegistry} registers {@link WrappingProxyCloseable}. When
+ * the proxy becomes subject to GC, this registry takes care of closing unclosed {@link Closeable}s.
+ * <p>
+ * Phantom references are used to track when {@link org.apache.flink.util.WrappingProxy}s of {@link Closeable} got
+ * GC'ed. We ensure that the wrapped {@link Closeable} is proeprly closed to avoid resource leaks.
+ * <p>
+ * Other than that, it works like a normal {@link CloseableRegistry}.
+ * <p>
+ * All methods in this class are thread-safe.
+ */
+public class SafetyNetCloseableRegistry extends
+ AbstractCloseableRegistry<WrappingProxyCloseable<? extends Closeable>,
+ SafetyNetCloseableRegistry.PhantomDelegatingCloseableRef> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SafetyNetCloseableRegistry.class);
+ private final ReferenceQueue<WrappingProxyCloseable<? extends Closeable>> referenceQueue;
+ private final Thread reaperThread;
+
+ public SafetyNetCloseableRegistry() {
+ super(new IdentityHashMap<Closeable, PhantomDelegatingCloseableRef>());
+ this.referenceQueue = new ReferenceQueue<>();
+ this.reaperThread = new Thread(new CloseableReaperRunnable());
+ reaperThread.start();
+ }
+
+ @Override
+ protected void doRegister(
+ WrappingProxyCloseable<? extends Closeable> wrappingProxyCloseable,
+ Map<Closeable, PhantomDelegatingCloseableRef> closeableMap) throws IOException {
+
+ Closeable innerCloseable = WrappingProxyUtil.stripProxy(wrappingProxyCloseable.getWrappedDelegate());
+
+ if (null == innerCloseable) {
+ return;
+ }
+
+ PhantomDelegatingCloseableRef phantomRef =
+ new PhantomDelegatingCloseableRef(wrappingProxyCloseable, referenceQueue);
+
+ closeableMap.put(innerCloseable, phantomRef);
+ }
+
+ @Override
+ protected void doUnRegister(
+ WrappingProxyCloseable<? extends Closeable> closeable,
+ Map<Closeable, PhantomDelegatingCloseableRef> closeableMap) {
+
+ Closeable innerCloseable = WrappingProxyUtil.stripProxy(closeable.getWrappedDelegate());
+
+ if (null == innerCloseable) {
+ return;
+ }
+
+ closeableMap.remove(innerCloseable);
+ }
+
+ /**
+ * Phantom reference to {@link WrappingProxyCloseable}.
+ */
+ static final class PhantomDelegatingCloseableRef
+ extends PhantomReference<WrappingProxyCloseable<? extends Closeable>>
+ implements Closeable {
+
+ private final Closeable innerCloseable;
+ private final String debugString;
+
+ public PhantomDelegatingCloseableRef(
+ WrappingProxyCloseable<? extends Closeable> referent,
+ ReferenceQueue<? super WrappingProxyCloseable<? extends Closeable>> q) {
+
+ super(referent, q);
+ this.innerCloseable = Preconditions.checkNotNull(WrappingProxyUtil.stripProxy(referent));
+ this.debugString = referent.toString();
+ }
+
+ public Closeable getInnerCloseable() {
+ return innerCloseable;
+ }
+
+ public String getDebugString() {
+ return debugString;
+ }
+
+ @Override
+ public void close() throws IOException {
+ innerCloseable.close();
+ }
+ }
+
+ /**
+ * Reaper runnable collects and closes leaking resources
+ */
+ final class CloseableReaperRunnable implements Runnable {
--- End diff --
I am usually against terminating threads purely via Interrupts. Its quite dangerous, because it only ever shows when the thread enters a blocking operation.
It will probably work here (not too much to do and frequently entering blocking calls), but to keep the code in the shape that it sets a good example for other contributors, lets add a proper `running` flag to control the `while` loop.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85330501
--- Diff: flink-core/src/main/java/org/apache/flink/util/IOUtils.java ---
@@ -213,6 +214,16 @@ public static void closeSocket(final Socket sock) {
}
}
}
+
+ public static void closeQuietly(Closeable closeable) {
+ try {
+ if (closeable != null) {
+ closeable.close();
+ }
+ } catch (IOException silent) {
--- End diff --
We typically call ignored exceptions `ignored` - that also makes IntelliJ not warn about ignoring them (if you have the inspection active)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2691: [FLINK-4910] Introduce safety net for closing file system...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/2691
Taking a look at this...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85330867
--- Diff: flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetWrapperFileSystem.java ---
@@ -0,0 +1,167 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingProxy;
+
+import java.io.IOException;
+import java.net.URI;
+
+/**
+ * This is a {@link WrappingProxy} around {@link FileSystem} which (i) wraps all opened streams as
+ * {@link ClosingFSDataInputStream} or {@link ClosingFSDataOutputStream} and (ii) registers them to
+ * a {@link SafetyNetCloseableRegistry}.
+ *
+ * Streams obtained by this are therefore managed by the {@link SafetyNetCloseableRegistry} to prevent resource leaks
+ * from unclosed streams.
+ */
+public class SafetyNetWrapperFileSystem extends FileSystem implements WrappingProxy<FileSystem> {
+
+ private final SafetyNetCloseableRegistry registry;
+ private final FileSystem unsafeFileSystem;
+
+ public SafetyNetWrapperFileSystem(FileSystem unsafeFileSystem, SafetyNetCloseableRegistry registry) {
+ this.registry = Preconditions.checkNotNull(registry);
+ this.unsafeFileSystem = Preconditions.checkNotNull(unsafeFileSystem);
+ }
+
+ public static FileSystem getLocalFileSystem() {
--- End diff --
Overriding the static methods is probably not necessary. I would not expect to use `SafetyNetWrapperFileSystem.get(...)`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85334057
--- Diff: flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java ---
@@ -55,6 +60,40 @@
private static final String HADOOP_WRAPPER_SCHEME = "hdwrapper";
+ private static final Logger LOG = LoggerFactory.getLogger(FileSystem.class);
+
+ /**
+ * Create a SafetyNetCloseableRegistry for a Task. This method should be called at the beginning of the task's
+ * main thread.
+ */
+ public static void createFileSystemCloseableRegistryForTask() {
+ SafetyNetCloseableRegistry current = REGISTRIES.get();
+ if (null != current) {
+ if (!current.isClosed()) {
+ IOUtils.closeQuietly(current);
+ LOG.warn("Found open SafetyNetCloseableRegistry. Closed and replaced it.");
+ }
+ }
+ REGISTRIES.set(new SafetyNetCloseableRegistry());
+ }
+
+ /**
+ * Create a SafetyNetCloseableRegistry for a Task. This method should be called at the end of the task's
+ * main thread or when the task should be canceled.
+ */
+ public static void disposeFileSystemCloseableRegistryForTask() {
+ SafetyNetCloseableRegistry ret = REGISTRIES.get();
+ if (null != ret) {
+ LOG.info("Closing SafetyNetCloseableRegistry.");
--- End diff --
Let's log something like "Ensuring all FileSystem streams are closed". Users will understand that better ;-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85331518
--- Diff: flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetCloseableRegistry.java ---
@@ -0,0 +1,165 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingProxyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
+import java.util.IdentityHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This implementation of an {@link AbstractCloseableRegistry} registers {@link WrappingProxyCloseable}. When
+ * the proxy becomes subject to GC, this registry takes care of closing unclosed {@link Closeable}s.
+ * <p>
+ * Phantom references are used to track when {@link org.apache.flink.util.WrappingProxy}s of {@link Closeable} got
+ * GC'ed. We ensure that the wrapped {@link Closeable} is proeprly closed to avoid resource leaks.
+ * <p>
+ * Other than that, it works like a normal {@link CloseableRegistry}.
+ * <p>
+ * All methods in this class are thread-safe.
+ */
+public class SafetyNetCloseableRegistry extends
+ AbstractCloseableRegistry<WrappingProxyCloseable<? extends Closeable>,
+ SafetyNetCloseableRegistry.PhantomDelegatingCloseableRef> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SafetyNetCloseableRegistry.class);
+ private final ReferenceQueue<WrappingProxyCloseable<? extends Closeable>> referenceQueue;
+ private final Thread reaperThread;
+
+ public SafetyNetCloseableRegistry() {
+ super(new IdentityHashMap<Closeable, PhantomDelegatingCloseableRef>());
+ this.referenceQueue = new ReferenceQueue<>();
+ this.reaperThread = new Thread(new CloseableReaperRunnable());
+ reaperThread.start();
--- End diff --
Let's make this thread a Daemon, and give it a good name (helps when dissecting Thread dumps)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2691: [FLINK-4910] Introduce safety net for closing file system...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/2691
Thanks for the reviews @StephanEwen and @aljoscha ! I addressed your comments in my last commit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85334345
--- Diff: flink-core/src/main/java/org/apache/flink/core/fs/AbstractCloseableRegistry.java ---
@@ -16,29 +16,29 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.state;
+package org.apache.flink.core.fs;
--- End diff --
Since this is used for other things than filesystems, how about moving it to `org.apache.flink.core.util`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2691: [FLINK-4910] Introduce safety net for closing file system...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/2691
R @StephanEwen and whoever is interested.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/2691#discussion_r85335033
--- Diff: flink-core/src/main/java/org/apache/flink/core/fs/SafetyNetCloseableRegistry.java ---
@@ -0,0 +1,165 @@
+/*
+ * 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.core.fs;
+
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingProxyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
+import java.util.IdentityHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This implementation of an {@link AbstractCloseableRegistry} registers {@link WrappingProxyCloseable}. When
+ * the proxy becomes subject to GC, this registry takes care of closing unclosed {@link Closeable}s.
+ * <p>
+ * Phantom references are used to track when {@link org.apache.flink.util.WrappingProxy}s of {@link Closeable} got
+ * GC'ed. We ensure that the wrapped {@link Closeable} is proeprly closed to avoid resource leaks.
+ * <p>
+ * Other than that, it works like a normal {@link CloseableRegistry}.
+ * <p>
+ * All methods in this class are thread-safe.
+ */
+public class SafetyNetCloseableRegistry extends
+ AbstractCloseableRegistry<WrappingProxyCloseable<? extends Closeable>,
+ SafetyNetCloseableRegistry.PhantomDelegatingCloseableRef> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SafetyNetCloseableRegistry.class);
+ private final ReferenceQueue<WrappingProxyCloseable<? extends Closeable>> referenceQueue;
+ private final Thread reaperThread;
+
+ public SafetyNetCloseableRegistry() {
+ super(new IdentityHashMap<Closeable, PhantomDelegatingCloseableRef>());
+ this.referenceQueue = new ReferenceQueue<>();
+ this.reaperThread = new Thread(new CloseableReaperRunnable());
+ reaperThread.start();
+ }
+
+ @Override
+ protected void doRegister(
+ WrappingProxyCloseable<? extends Closeable> wrappingProxyCloseable,
+ Map<Closeable, PhantomDelegatingCloseableRef> closeableMap) throws IOException {
+
+ Closeable innerCloseable = WrappingProxyUtil.stripProxy(wrappingProxyCloseable.getWrappedDelegate());
+
+ if (null == innerCloseable) {
+ return;
+ }
+
+ PhantomDelegatingCloseableRef phantomRef =
+ new PhantomDelegatingCloseableRef(wrappingProxyCloseable, referenceQueue);
+
+ closeableMap.put(innerCloseable, phantomRef);
+ }
+
+ @Override
+ protected void doUnRegister(
+ WrappingProxyCloseable<? extends Closeable> closeable,
+ Map<Closeable, PhantomDelegatingCloseableRef> closeableMap) {
+
+ Closeable innerCloseable = WrappingProxyUtil.stripProxy(closeable.getWrappedDelegate());
+
+ if (null == innerCloseable) {
+ return;
+ }
+
+ closeableMap.remove(innerCloseable);
+ }
+
+ /**
+ * Phantom reference to {@link WrappingProxyCloseable}.
+ */
+ static final class PhantomDelegatingCloseableRef
+ extends PhantomReference<WrappingProxyCloseable<? extends Closeable>>
+ implements Closeable {
+
+ private final Closeable innerCloseable;
+ private final String debugString;
+
+ public PhantomDelegatingCloseableRef(
+ WrappingProxyCloseable<? extends Closeable> referent,
+ ReferenceQueue<? super WrappingProxyCloseable<? extends Closeable>> q) {
+
+ super(referent, q);
+ this.innerCloseable = Preconditions.checkNotNull(WrappingProxyUtil.stripProxy(referent));
+ this.debugString = referent.toString();
+ }
+
+ public Closeable getInnerCloseable() {
+ return innerCloseable;
+ }
+
+ public String getDebugString() {
+ return debugString;
+ }
+
+ @Override
+ public void close() throws IOException {
+ innerCloseable.close();
+ }
+ }
+
+ /**
+ * Reaper runnable collects and closes leaking resources
+ */
+ final class CloseableReaperRunnable implements Runnable {
+ @Override
+ public void run() {
+ try {
+ List<PhantomDelegatingCloseableRef> closeableList = new LinkedList<>();
--- End diff --
Isn't an array list more efficient here, especially if you reuse the list?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2691: [FLINK-4910] Introduce safety net for closing file...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/2691
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2691: [FLINK-4910] Introduce safety net for closing file system...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2691
Thanks for the great contribution @StefanRRichter. Merging this PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---