[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15303884#comment-15303884 ] Pawel Szczur commented on FLINK-3291: - I've browsed through a Flink issues to find anything related to this: http://stackoverflow.com/questions/37473682/items-not-groupped-correctly-cogroupbykey Basing on the description and some comments, is it possible my problem is related to this bug? > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15303994#comment-15303994 ] Gabor Gevay commented on FLINK-3291: This bug occurred when object reuse is enabled. This is disabled by default, and can be enabled by calling ExecutionConfig.enableObjectReuse. I don't know Beam, but I would guess that this option is also not enabled there by default. > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[
https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15304234#comment-15304234
]
Greg Hogan commented on FLINK-3291:
---
The {{inputStream}} field causing the {{IllegalArgumentException}} was changed
from package-private to protected in [0]. Could your system be using an older
version of Kryo?
[0]
https://github.com/EsotericSoftware/kryo/commit/642972d532e832aabfe8c9be515b8b2560f093ce#diff-d6b5559e1180e131a85fa781a469f0b2L19
> Object reuse bug in MergeIterator.HeadStream.nextHead
> -
>
> Key: FLINK-3291
> URL: https://issues.apache.org/jira/browse/FLINK-3291
> Project: Flink
> Issue Type: Bug
> Components: Distributed Runtime
>Affects Versions: 1.0.0
>Reporter: Gabor Gevay
>Assignee: Gabor Gevay
>Priority: Critical
>
> MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the
> `reuse` object that it got as an argument. This object might be modified
> later by the caller.
> This actually happens when ReduceDriver.run calls input.next (which will
> actually be MergeIterator.next(E reuse)) in the inner while loop of the
> objectReuseEnabled branch, and that calls top.nextHead with the reference
> that it got from ReduceDriver, which erroneously saves the reference, and
> then ReduceDriver later uses that same object for doing the reduce.
> Another way in which this fails is when MergeIterator.next(E reuse) gives
> `reuse` to different `top`s in different calls, and then the heads end up
> being the same object.
> You can observe the latter situation in action by running ReducePerformance
> here:
> https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug
> Set memory to -Xmx200m (so that the MergeIterator actually has merging to
> do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then
> watch `reuse`, and the heads of the first two elements of `this.heap` in the
> debugger. They will get to be the same object after hitting continue about 6
> times.
> You can also look at the count that is printed at the end, which shouldn't be
> larger than the key range. Also, if you look into the output file
> /tmp/xxxobjectreusebug, for example the key 77 appears twice.
> The good news is that I think I can see an easy fix that doesn't affect
> performance: MergeIterator.HeadStream could have a reuse object of its own as
> a member, and give that to iterator.next in nextHead(E reuse). And then we
> wouldn't need the overload of nextHead that has the reuse parameter, and
> MergeIterator.next(E reuse) could just call its other overload.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15307687#comment-15307687 ] Pawel Szczur commented on FLINK-3291: - [~ggevay] Thanks for explanation. [~greghogan] I cannot reproduce it. The version of Kryo used is 2.24, released in May 2014. The discussion about this problem in Flink backend for Beam: http://mail-archives.apache.org/mod_mbox/incubator-beam-user/201605.mbox/%3CCAB2uKkG2xHsWpLFUkYnt8eEzdxU%3DB_nu6crTwVi-ZuUpugxkPQ%40mail.gmail.com%3E It seems that GroupByKey is broken. Beam is a facade API for a data processing frameworks. > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[
https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15128219#comment-15128219
]
Greg Hogan commented on FLINK-3291:
---
Hi Gabor, this looks to be a bug in some code I had touched recently.
`ReduceDriver.run` owns two objects but is not tracking the object returned in
`while ((value = input.next(reuse2)) != null) {`. If you aren't already working
on this, I'd like to have a go at reworking this.
> Object reuse bug in MergeIterator.HeadStream.nextHead
> -
>
> Key: FLINK-3291
> URL: https://issues.apache.org/jira/browse/FLINK-3291
> Project: Flink
> Issue Type: Bug
> Components: Distributed Runtime
>Affects Versions: 1.0.0
>Reporter: Gabor Gevay
>Assignee: Gabor Gevay
>Priority: Critical
>
> MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the
> `reuse` object that it got as an argument. This object might be modified
> later by the caller.
> This actually happens when ReduceDriver.run calls input.next (which will
> actually be MergeIterator.next(E reuse)) in the inner while loop of the
> objectReuseEnabled branch, and that calls top.nextHead with the reference
> that it got from ReduceDriver, which erroneously saves the reference, and
> then ReduceDriver later uses that same object for doing the reduce.
> Another way in which this fails is when MergeIterator.next(E reuse) gives
> `reuse` to different `top`s in different calls, and then the heads end up
> being the same object.
> You can observe the latter situation in action by running ReducePerformance
> here:
> https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug
> Set memory to -Xmx200m (so that the MergeIterator actually has merging to
> do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then
> watch `reuse`, and the heads of the first two elements of `this.heap` in the
> debugger. They will get to be the same object after hitting continue about 6
> times.
> You can also look at the count that is printed at the end, which shouldn't be
> larger than the key range. Also, if you look into the output file
> /tmp/xxxobjectreusebug, for example the key 77 appears twice.
> The good news is that I think I can see an easy fix that doesn't affect
> performance: MergeIterator.HeadStream could have a reuse object of its own as
> a member, and give that to iterator.next in nextHead(E reuse). And then we
> wouldn't need the overload of nextHead that has the reuse parameter, and
> MergeIterator.next(E reuse) could just call its other overload.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15128225#comment-15128225 ] Gabor Gevay commented on FLINK-3291: I'm almost done with the fix, see here: https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug-fix1 The only thing that is missing is adding a small test for specifically this problem. And I would like to additionally figure out why the test that are already in place didn't catch this problem. Btw. I couldn't avoid introducing an extra copy. (Th solution that I wrote in the Jira description, doesn't really work in itself.) > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[
https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15128243#comment-15128243
]
Gabor Gevay commented on FLINK-3291:
> `ReduceDriver.run` owns two objects but is not tracking the object returned
> in `while ((value = input.next(reuse2)) != null) {`
The problem is more complicated than that: This can't be done with just 1 or 2
reuse objects, each of the head streams need to own objects themselves.
> Object reuse bug in MergeIterator.HeadStream.nextHead
> -
>
> Key: FLINK-3291
> URL: https://issues.apache.org/jira/browse/FLINK-3291
> Project: Flink
> Issue Type: Bug
> Components: Distributed Runtime
>Affects Versions: 1.0.0
>Reporter: Gabor Gevay
>Assignee: Gabor Gevay
>Priority: Critical
>
> MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the
> `reuse` object that it got as an argument. This object might be modified
> later by the caller.
> This actually happens when ReduceDriver.run calls input.next (which will
> actually be MergeIterator.next(E reuse)) in the inner while loop of the
> objectReuseEnabled branch, and that calls top.nextHead with the reference
> that it got from ReduceDriver, which erroneously saves the reference, and
> then ReduceDriver later uses that same object for doing the reduce.
> Another way in which this fails is when MergeIterator.next(E reuse) gives
> `reuse` to different `top`s in different calls, and then the heads end up
> being the same object.
> You can observe the latter situation in action by running ReducePerformance
> here:
> https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug
> Set memory to -Xmx200m (so that the MergeIterator actually has merging to
> do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then
> watch `reuse`, and the heads of the first two elements of `this.heap` in the
> debugger. They will get to be the same object after hitting continue about 6
> times.
> You can also look at the count that is printed at the end, which shouldn't be
> larger than the key range. Also, if you look into the output file
> /tmp/xxxobjectreusebug, for example the key 77 appears twice.
> The good news is that I think I can see an easy fix that doesn't affect
> performance: MergeIterator.HeadStream could have a reuse object of its own as
> a member, and give that to iterator.next in nextHead(E reuse). And then we
> wouldn't need the overload of nextHead that has the reuse parameter, and
> MergeIterator.next(E reuse) could just call its other overload.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15130452#comment-15130452 ] Gabor Gevay commented on FLINK-3291: OK, sorry, I think I have now understood what you meant by ReduceDriver.run not tracking the object returned from the iterator call. The problem here is that after 0a8df6d513fa59d650ff875bdf3a1613d0f14af5, I mustn't modify an object that I have given to an iterator.next call as a reuse object, because MergeIterator.HeadStream.nextHead saves a reference to it, and expects that object to not change. But this seems like a rather scary requirement, and I wouldn't be sure that some other code besides ReduceDriver somewhere doesn't also violate it. I think that the root cause of these issues, is that the documentation about object reuse [1] is rather inadequate in clearly stating what are the contracts in this area, so I tried to put together a Google Doc about this: [2]. Could you please look at it, and tell me how much it aligns with your way of thinking about object reuse? [1] https://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html#object-reuse-behavior [2] https://docs.google.com/document/d/1cgkuttvmj4jUonG7E2RdFVjKlfQDm_hE6gvFcgAfzXg/edit > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15131222#comment-15131222 ] Greg Hogan commented on FLINK-3291: --- Object reuse is active when object reuse is enabled or when objects are passed between chained functions. When object reuse is active a user may modify any non-keyed fields but should not access function parameters across function calls. The only prohibitions on users are: 1) never change keyed fields when returning an object 2) copy mutable objects retained between function calls unless a) object reuse is disabled and 2) the operator is not chainable That's my understanding. > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15134152#comment-15134152 ] Gabor Gevay commented on FLINK-3291: @Greg, my current problem is that I'm not sure what should be the exact contract for `MutableObjectIterator.next(reuse)`. The fix that I have so far \[1\] makes 3b "yes" (see \[2\]), but it seems that you would rather go for "no", so that it is possible to handle object reuse by "exchanging the given object for an existing object" as you say in your comment \[3\]. Do you think that the performance gain (avoiding a copy) makes the complications that arise at every use of `MutableObjectIterator.next(reuse)` worthwhile? \[1\] https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug-fix1 \[2\] https://docs.google.com/document/d/1cgkuttvmj4jUonG7E2RdFVjKlfQDm_hE6gvFcgAfzXg/edit \[3\] https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/MergeIterator.java#L70 > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[
https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15144654#comment-15144654
]
Stephan Ewen commented on FLINK-3291:
-
Thanks, @Greg and [~ggevay] for looking into this. It's good some people start
going in-depth there.
Here are a few comments from what I thought initially when we implemented the
first object reuse versions:
- The initial idea of a contract for {{MutableObjectiterator.next(reuse)}}
was the following:
1. The caller may not hold onto {{reuse}} any more
2. The iterator implementor may not hold onto the returned object any
more.
Given that this was long ago (5 years probably, since I created that
interface), I am pretty sure that contract is not obeyed everywhere.
- In all cases, the non-reuse implementations should work without
{{serializer.createInstance()}}, because in some corner cases, neither Flink
nor Kryo/Objenesis manages to instantiate the object.
- Copying objects is sometimes necessary, but should be avoided where
possible. In microbenchmarks, the copies are often cheap (tuples of Strings,
Integers), in practice many objects are JSON, Avro, Thrift, and are hellishly
expensive to copy.
Especially in the non-reusing mode, I should be completely avoidable. The
object reuse-mode is mainly for programs with types that are mutable and
efficient to copy. Reuse can become more expensive the non-reuse if copies are
expensive.
> Object reuse bug in MergeIterator.HeadStream.nextHead
> -
>
> Key: FLINK-3291
> URL: https://issues.apache.org/jira/browse/FLINK-3291
> Project: Flink
> Issue Type: Bug
> Components: Distributed Runtime
>Affects Versions: 1.0.0
>Reporter: Gabor Gevay
>Assignee: Gabor Gevay
>Priority: Critical
>
> MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the
> `reuse` object that it got as an argument. This object might be modified
> later by the caller.
> This actually happens when ReduceDriver.run calls input.next (which will
> actually be MergeIterator.next(E reuse)) in the inner while loop of the
> objectReuseEnabled branch, and that calls top.nextHead with the reference
> that it got from ReduceDriver, which erroneously saves the reference, and
> then ReduceDriver later uses that same object for doing the reduce.
> Another way in which this fails is when MergeIterator.next(E reuse) gives
> `reuse` to different `top`s in different calls, and then the heads end up
> being the same object.
> You can observe the latter situation in action by running ReducePerformance
> here:
> https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug
> Set memory to -Xmx200m (so that the MergeIterator actually has merging to
> do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then
> watch `reuse`, and the heads of the first two elements of `this.heap` in the
> debugger. They will get to be the same object after hitting continue about 6
> times.
> You can also look at the count that is printed at the end, which shouldn't be
> larger than the key range. Also, if you look into the output file
> /tmp/xxxobjectreusebug, for example the key 77 appears twice.
> The good news is that I think I can see an easy fix that doesn't affect
> performance: MergeIterator.HeadStream could have a reuse object of its own as
> a member, and give that to iterator.next in nextHead(E reuse). And then we
> wouldn't need the overload of nextHead that has the reuse parameter, and
> MergeIterator.next(E reuse) could just call its other overload.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15144732#comment-15144732 ] Greg Hogan commented on FLINK-3291: --- I started looking at `MergeIterator` last year since I had object reuse enabled and was still seeing a flood of newly created objects. The complex code was already written, but tying it together and passing the configuration through resulted in ~5% reduction in overall runtime for the algorithms I was benchmarking. So clearly this is a performance critical section of the code, and as evidenced by this ticket a source of potential errors. [~ggevay], at this point are the changes to `MergeIterator` fixing a bug? Do you want to fix up and clarify the documentation for `MutableObjectIterator` and verify the implementing classes? My only question on [~StephanEwen]'s comments would be, isn't copying sometimes necessary in non-reusing mode but avoidable when reusing objects? Copies would be required any time an object is presented to the user multiple times, as in joins / cross / cogroup. [~StephanEwen]'s comment on cheap vs expensive copies would be excellent for the user documentation. As I understand, if you're making expensive copies (with object reuse enabled) in a user defined function, you may be better off disabling object reuse and allowing the deserialization to use a fresh object instead. > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[
https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15144755#comment-15144755
]
Gabor Gevay commented on FLINK-3291:
{quote}
The initial idea of a contract for MutableObjectiterator.next(reuse) was the
following:
1. The caller may not hold onto reuse any more
2. The iterator implementor may not hold onto the returned object any more.
Given that this was long ago (5 years probably, since I created that
interface), I am pretty sure that contract is not obeyed everywhere.
{quote}
OK, this clears things up; thanks for chiming in, [~StephanEwen]!
Then the things we should do are
1. Add this contract to the javadoc of `MutableObjectiterator.next(reuse)`.
2. Go with [~greghogan]'s solution \[1\] to fix the problem brought up by this
Jira.
3. Check all calls to `MutableObjectiterator.next(reuse)`. (A few suspicious
ones are in `CrossDriver`, `UnionWithTempOperator`,
`MutableHashTable.ProbeIterator.next`, and
`ReusingBuildFirstHashJoinIterator.callWithNextKey`.)
{quote}
Gabor Gevay, at this point are the changes to MergeIterator fixing a bug? Do
you want to fix up and clarify the documentation for MutableObjectIterator and
verify the implementing classes?
{quote}
My changes to `MergeIterator` become unnecessary if the contracts of
`MutableObjectIterator.next(reuse)` are what Stephan said.
I'll open a Jira for fixing the javadoc of `MutableObjectIterator.next(reuse)`
to include the contract (and possibly also add this info to a new wiki page
that will be linked from the new version of the object reuse documentation
brewing at \[2\]), and check all call sites whether they obey the contract.
\[1\] https://github.com/apache/flink/pull/1626
\[2\]
https://docs.google.com/document/d/1cgkuttvmj4jUonG7E2RdFVjKlfQDm_hE6gvFcgAfzXg/edit
> Object reuse bug in MergeIterator.HeadStream.nextHead
> -
>
> Key: FLINK-3291
> URL: https://issues.apache.org/jira/browse/FLINK-3291
> Project: Flink
> Issue Type: Bug
> Components: Distributed Runtime
>Affects Versions: 1.0.0
>Reporter: Gabor Gevay
>Assignee: Gabor Gevay
>Priority: Critical
>
> MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the
> `reuse` object that it got as an argument. This object might be modified
> later by the caller.
> This actually happens when ReduceDriver.run calls input.next (which will
> actually be MergeIterator.next(E reuse)) in the inner while loop of the
> objectReuseEnabled branch, and that calls top.nextHead with the reference
> that it got from ReduceDriver, which erroneously saves the reference, and
> then ReduceDriver later uses that same object for doing the reduce.
> Another way in which this fails is when MergeIterator.next(E reuse) gives
> `reuse` to different `top`s in different calls, and then the heads end up
> being the same object.
> You can observe the latter situation in action by running ReducePerformance
> here:
> https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug
> Set memory to -Xmx200m (so that the MergeIterator actually has merging to
> do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then
> watch `reuse`, and the heads of the first two elements of `this.heap` in the
> debugger. They will get to be the same object after hitting continue about 6
> times.
> You can also look at the count that is printed at the end, which shouldn't be
> larger than the key range. Also, if you look into the output file
> /tmp/xxxobjectreusebug, for example the key 77 appears twice.
> The good news is that I think I can see an easy fix that doesn't affect
> performance: MergeIterator.HeadStream could have a reuse object of its own as
> a member, and give that to iterator.next in nextHead(E reuse). And then we
> wouldn't need the overload of nextHead that has the reuse parameter, and
> MergeIterator.next(E reuse) could just call its other overload.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15144821#comment-15144821 ] Stephan Ewen commented on FLINK-3291: - Just saying that this contract is not carved in stone. > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[
https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15144823#comment-15144823
]
Gabor Gevay commented on FLINK-3291:
I tried to do that, but with the one I came up with, I couldn't avoid
introducing an extra copy in {{MergeIterator.HeadStream.getHead}}.
> Object reuse bug in MergeIterator.HeadStream.nextHead
> -
>
> Key: FLINK-3291
> URL: https://issues.apache.org/jira/browse/FLINK-3291
> Project: Flink
> Issue Type: Bug
> Components: Distributed Runtime
>Affects Versions: 1.0.0
>Reporter: Gabor Gevay
>Assignee: Gabor Gevay
>Priority: Critical
>
> MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the
> `reuse` object that it got as an argument. This object might be modified
> later by the caller.
> This actually happens when ReduceDriver.run calls input.next (which will
> actually be MergeIterator.next(E reuse)) in the inner while loop of the
> objectReuseEnabled branch, and that calls top.nextHead with the reference
> that it got from ReduceDriver, which erroneously saves the reference, and
> then ReduceDriver later uses that same object for doing the reduce.
> Another way in which this fails is when MergeIterator.next(E reuse) gives
> `reuse` to different `top`s in different calls, and then the heads end up
> being the same object.
> You can observe the latter situation in action by running ReducePerformance
> here:
> https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug
> Set memory to -Xmx200m (so that the MergeIterator actually has merging to
> do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then
> watch `reuse`, and the heads of the first two elements of `this.heap` in the
> debugger. They will get to be the same object after hitting continue about 6
> times.
> You can also look at the count that is printed at the end, which shouldn't be
> larger than the key range. Also, if you look into the output file
> /tmp/xxxobjectreusebug, for example the key 77 appears twice.
> The good news is that I think I can see an easy fix that doesn't affect
> performance: MergeIterator.HeadStream could have a reuse object of its own as
> a member, and give that to iterator.next in nextHead(E reuse). And then we
> wouldn't need the overload of nextHead that has the reuse parameter, and
> MergeIterator.next(E reuse) could just call its other overload.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (FLINK-3291) Object reuse bug in MergeIterator.HeadStream.nextHead
[ https://issues.apache.org/jira/browse/FLINK-3291?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15144815#comment-15144815 ] Stephan Ewen commented on FLINK-3291: - What I quoted there was the initial contract. It may actually be that you guys come up with a better contract... > Object reuse bug in MergeIterator.HeadStream.nextHead > - > > Key: FLINK-3291 > URL: https://issues.apache.org/jira/browse/FLINK-3291 > Project: Flink > Issue Type: Bug > Components: Distributed Runtime >Affects Versions: 1.0.0 >Reporter: Gabor Gevay >Assignee: Gabor Gevay >Priority: Critical > > MergeIterator.HeadStream.nextHead saves a reference into `this.head` of the > `reuse` object that it got as an argument. This object might be modified > later by the caller. > This actually happens when ReduceDriver.run calls input.next (which will > actually be MergeIterator.next(E reuse)) in the inner while loop of the > objectReuseEnabled branch, and that calls top.nextHead with the reference > that it got from ReduceDriver, which erroneously saves the reference, and > then ReduceDriver later uses that same object for doing the reduce. > Another way in which this fails is when MergeIterator.next(E reuse) gives > `reuse` to different `top`s in different calls, and then the heads end up > being the same object. > You can observe the latter situation in action by running ReducePerformance > here: > https://github.com/ggevay/flink/tree/merge-iterator-object-reuse-bug > Set memory to -Xmx200m (so that the MergeIterator actually has merging to > do), put a breakpoint at the beginning of MergeIterator.next(reuse), and then > watch `reuse`, and the heads of the first two elements of `this.heap` in the > debugger. They will get to be the same object after hitting continue about 6 > times. > You can also look at the count that is printed at the end, which shouldn't be > larger than the key range. Also, if you look into the output file > /tmp/xxxobjectreusebug, for example the key 77 appears twice. > The good news is that I think I can see an easy fix that doesn't affect > performance: MergeIterator.HeadStream could have a reuse object of its own as > a member, and give that to iterator.next in nextHead(E reuse). And then we > wouldn't need the overload of nextHead that has the reuse parameter, and > MergeIterator.next(E reuse) could just call its other overload. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
