You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by aljoscha <gi...@git.apache.org> on 2014/12/15 19:53:02 UTC
[GitHub] incubator-flink pull request: [FLINK-1325] Add Java ClosureCleaner
GitHub user aljoscha opened a pull request:
https://github.com/apache/incubator-flink/pull/269
[FLINK-1325] Add Java ClosureCleaner
This also adds the ExecutionConfig, where the user can enable/disable the closure cleaner and other settings. I'm also planning to use this for the object-reuse settings in the future.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/aljoscha/incubator-flink java-closure-cleaner
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-flink/pull/269.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 #269
----
commit 685da9e47b89a56e5f900d1a7f9b5ab77bc51c67
Author: Aljoscha Krettek <al...@gmail.com>
Date: 2014-12-15T17:37:42Z
Add ExecutionConfig
commit fe51a9c50963bb5c0a6fd0beb8db57c94685adbc
Author: Aljoscha Krettek <al...@gmail.com>
Date: 2014-12-15T18:50:14Z
[FLINK-1325] Add Java ClosureCleaner
----
---
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] incubator-flink pull request: [FLINK-1325] Add Java ClosureCleaner
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/incubator-flink/pull/269#issuecomment-67220504
Looks good, I'll merge it with the current bunch...
---
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] incubator-flink pull request: [FLINK-1325] Add Java ClosureCleaner
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/269#discussion_r21900060
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/ClosureCleaner.java ---
@@ -0,0 +1,131 @@
+/*
+ * 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.api.java;
+
+
+import org.apache.flink.api.common.InvalidProgramException;
+import org.apache.flink.util.InstantiationUtil;
+import org.objectweb.asm.ClassReader;
+import org.objectweb.asm.ClassVisitor;
+import org.objectweb.asm.MethodVisitor;
+import org.objectweb.asm.Opcodes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+
+public class ClosureCleaner {
+ private static Logger LOG = LoggerFactory.getLogger(ClosureCleaner.class);
+
+ private static ClassReader getClassReader(Class<?> cls) {
+ String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
+ try {
+ return new ClassReader(cls.getResourceAsStream(className));
+ } catch (IOException e) {
+ throw new RuntimeException("Could not create ClassReader: " + e);
+ }
+ }
+
+ public static void clean(Object func, boolean checkSerializable) {
+ Class<?> cls = func.getClass();
+
+ String this0Name = null;
+
+ // First find the field name of the "this$0" field, this can
+ // be "field$x" depending on the nesting
+ for (Field f: cls.getDeclaredFields()) {
+ if (f.getName().startsWith("this$")) {
+ // found our field:
+ this0Name = f.getName();
--- End diff --
Is it intentional to only check the last field (there is also no beak)? Should this check all fields with this pattern?
---
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] incubator-flink pull request: [FLINK-1325] Add Java ClosureCleaner
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/incubator-flink/pull/269#issuecomment-67168154
The cleaner visits all methods, yes. It does not know about how we use those functions. I'll change one of the tests to access the outer object through another method.
---
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] incubator-flink pull request: [FLINK-1325] Add Java ClosureCleaner
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-flink/pull/269
---
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] incubator-flink pull request: [FLINK-1325] Add Java ClosureCleaner
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/incubator-flink/pull/269#issuecomment-67166915
Looks good.
Here are some issues that we need to watch out for:
- I think we need to have the ASM dependency as a managed dependency in the parent pom. There are different parts that use ASM in potentially different versions (the JAR builder in a pending pull request uses ASM 5 to support lambdas)
- Does the closure cleaner visit all methods? (I am not familiar enough with ASM to figure that out from the code)
- Should we add a test where the outer field is accessed not from the `map()` method itself, but from another method?
---
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] incubator-flink pull request: [FLINK-1325] Add Java ClosureCleaner
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/269#discussion_r21900493
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/ClosureCleaner.java ---
@@ -0,0 +1,131 @@
+/*
+ * 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.api.java;
+
+
+import org.apache.flink.api.common.InvalidProgramException;
+import org.apache.flink.util.InstantiationUtil;
+import org.objectweb.asm.ClassReader;
+import org.objectweb.asm.ClassVisitor;
+import org.objectweb.asm.MethodVisitor;
+import org.objectweb.asm.Opcodes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+
+public class ClosureCleaner {
+ private static Logger LOG = LoggerFactory.getLogger(ClosureCleaner.class);
+
+ private static ClassReader getClassReader(Class<?> cls) {
+ String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
+ try {
+ return new ClassReader(cls.getResourceAsStream(className));
+ } catch (IOException e) {
+ throw new RuntimeException("Could not create ClassReader: " + e);
+ }
+ }
+
+ public static void clean(Object func, boolean checkSerializable) {
+ Class<?> cls = func.getClass();
+
+ String this0Name = null;
+
+ // First find the field name of the "this$0" field, this can
+ // be "field$x" depending on the nesting
+ for (Field f: cls.getDeclaredFields()) {
+ if (f.getName().startsWith("this$")) {
+ // found our field:
+ this0Name = f.getName();
--- End diff --
It could be that only one field of the pattern is allowed. Just to be on the safe side, I'm changing it to check all fields of the pattern "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] incubator-flink pull request: [FLINK-1325] Add Java ClosureCleaner
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/269#discussion_r21900264
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/ClosureCleaner.java ---
@@ -0,0 +1,131 @@
+/*
+ * 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.api.java;
+
+
+import org.apache.flink.api.common.InvalidProgramException;
+import org.apache.flink.util.InstantiationUtil;
+import org.objectweb.asm.ClassReader;
+import org.objectweb.asm.ClassVisitor;
+import org.objectweb.asm.MethodVisitor;
+import org.objectweb.asm.Opcodes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+
+public class ClosureCleaner {
+ private static Logger LOG = LoggerFactory.getLogger(ClosureCleaner.class);
+
+ private static ClassReader getClassReader(Class<?> cls) {
+ String className = cls.getName().replaceFirst("^.*\\.", "") + ".class";
+ try {
+ return new ClassReader(cls.getResourceAsStream(className));
+ } catch (IOException e) {
+ throw new RuntimeException("Could not create ClassReader: " + e);
+ }
+ }
+
+ public static void clean(Object func, boolean checkSerializable) {
+ Class<?> cls = func.getClass();
+
+ String this0Name = null;
+
+ // First find the field name of the "this$0" field, this can
+ // be "field$x" depending on the nesting
+ for (Field f: cls.getDeclaredFields()) {
+ if (f.getName().startsWith("this$")) {
+ // found our field:
+ this0Name = f.getName();
+ }
+ }
+
+ if (this0Name == null) {
+ // no this$0 field, just return
+ return;
+ }
+
+ This0AccessFinder this0Finder = new This0AccessFinder(this0Name);
+
+ getClassReader(cls).accept(this0Finder, 0);
+
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(this0Name + " is accessed: " + this0Finder.isThis0Accessed());
+ }
+
+ if (!this0Finder.isThis0Accessed()) {
+ Field this0;
+ try {
+ this0 = func.getClass().getDeclaredField(this0Name);
+ } catch (NoSuchFieldException e) {
+ // has no this$0, just return
+ throw new RuntimeException("Could not set this$0: " + e);
+ }
+ this0.setAccessible(true);
+ try {
+ this0.set(func, null);
+ } catch (IllegalAccessException e) {
+ // should not happen, since we use setAccessible
+ throw new RuntimeException("Could not set this$0: " + e);
+ }
+ }
+
+ if (checkSerializable) {
+ ensureSerializable(func);
+ }
+ }
+
+ public static void ensureSerializable(Object func) {
+ try {
+ InstantiationUtil.serializeObject(func);
--- End diff --
That is probably okay for now, but for objects with large closures, we are doing significant double work...
---
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.
---